<!DOCTYPE html>
<html lang="en-US">
  <head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width,initial-scale=1">
    <title>KafkaProducer使用介绍、参数配置以及核心源码解析 | JavaKeeper</title>
    <meta name="generator" content="VuePress 1.5.4">
    <link rel="icon" href="/icon.svg">
    <script>
        var _hmt = _hmt || [];
        (function() {
            var hm = document.createElement("script");
            hm.src = "https://hm.baidu.com/hm.js?a949a9b30eb86ac0159e735ff8670c03";
            var s = document.getElementsByTagName("script")[0];
            s.parentNode.insertBefore(hm, s);
            // 引入谷歌,不需要可删除这段
            var hm1 = document.createElement("script");
            hm1.src = "https://www.googletagmanager.com/gtag/js?id=UA-169923503-1";
            var s1 = document.getElementsByTagName("script")[0]; 
            s1.parentNode.insertBefore(hm1, s1);
        })();
        // 谷歌加载,不需要可删除
        window.dataLayer = window.dataLayer || [];
        function gtag(){dataLayer.push(arguments);}
        gtag('js', new Date());
        gtag('config', 'UA-169923503-1');
    </script>
    <meta name="description" content="">
    <meta name="keywords" content="JavaKeeper,Java,Java开发,算法,blog">
    <link rel="preload" href="/assets/css/0.styles.91f57736.css" as="style"><link rel="preload" href="/assets/js/app.447d4224.js" as="script"><link rel="preload" href="/assets/js/3.9d76740c.js" as="script"><link rel="preload" href="/assets/js/1.c4fd7d2e.js" as="script"><link rel="preload" href="/assets/js/154.e090b491.js" as="script"><link rel="prefetch" href="/assets/js/10.8cf3be2c.js"><link rel="prefetch" href="/assets/js/100.74f35ab8.js"><link rel="prefetch" href="/assets/js/101.7a062346.js"><link rel="prefetch" href="/assets/js/102.c9485235.js"><link rel="prefetch" href="/assets/js/103.d88a3805.js"><link rel="prefetch" href="/assets/js/104.6e034144.js"><link rel="prefetch" href="/assets/js/105.d22f7450.js"><link rel="prefetch" href="/assets/js/106.a6cb54b0.js"><link rel="prefetch" href="/assets/js/107.7b65e72b.js"><link rel="prefetch" href="/assets/js/108.eb5804bb.js"><link rel="prefetch" href="/assets/js/109.05f775e5.js"><link rel="prefetch" href="/assets/js/11.c54ae13c.js"><link rel="prefetch" href="/assets/js/110.51d3d641.js"><link rel="prefetch" href="/assets/js/111.022b64a7.js"><link rel="prefetch" href="/assets/js/112.da8afd52.js"><link rel="prefetch" href="/assets/js/113.05a17b18.js"><link rel="prefetch" href="/assets/js/114.8960d913.js"><link rel="prefetch" href="/assets/js/115.67919f68.js"><link rel="prefetch" href="/assets/js/116.62b0cd71.js"><link rel="prefetch" href="/assets/js/117.ebac3eff.js"><link rel="prefetch" href="/assets/js/118.ecd629bd.js"><link rel="prefetch" href="/assets/js/119.a09a0897.js"><link rel="prefetch" href="/assets/js/12.60aa3b24.js"><link rel="prefetch" href="/assets/js/120.bf639d3d.js"><link rel="prefetch" href="/assets/js/121.b89d0c8e.js"><link rel="prefetch" href="/assets/js/122.1a75ff83.js"><link rel="prefetch" href="/assets/js/123.d2127132.js"><link rel="prefetch" href="/assets/js/124.2caff9e0.js"><link rel="prefetch" href="/assets/js/125.9b9f966a.js"><link rel="prefetch" href="/assets/js/126.58cdfb3d.js"><link rel="prefetch" href="/assets/js/127.8ef09c53.js"><link rel="prefetch" href="/assets/js/128.efdc2ae4.js"><link rel="prefetch" href="/assets/js/129.e35cbc57.js"><link rel="prefetch" href="/assets/js/13.125c13a0.js"><link rel="prefetch" href="/assets/js/130.f01a55e3.js"><link rel="prefetch" href="/assets/js/131.65205f4a.js"><link rel="prefetch" href="/assets/js/132.f42c5a0a.js"><link rel="prefetch" href="/assets/js/133.9ba468b3.js"><link rel="prefetch" href="/assets/js/134.7b597ba9.js"><link rel="prefetch" href="/assets/js/135.fb828b9a.js"><link rel="prefetch" href="/assets/js/136.3887532f.js"><link rel="prefetch" href="/assets/js/137.549bae01.js"><link rel="prefetch" href="/assets/js/138.db8d423d.js"><link rel="prefetch" href="/assets/js/139.dbaf2267.js"><link rel="prefetch" href="/assets/js/14.bd1d0b0d.js"><link rel="prefetch" href="/assets/js/140.6cb65fdc.js"><link rel="prefetch" href="/assets/js/141.9bd6cc4b.js"><link rel="prefetch" href="/assets/js/142.552db5ed.js"><link rel="prefetch" href="/assets/js/143.2c9f2bf4.js"><link rel="prefetch" href="/assets/js/144.fba98a15.js"><link rel="prefetch" href="/assets/js/145.c42f3a21.js"><link rel="prefetch" href="/assets/js/146.596d4d33.js"><link rel="prefetch" href="/assets/js/147.c48ae5c1.js"><link rel="prefetch" href="/assets/js/148.71064871.js"><link rel="prefetch" href="/assets/js/149.16582d21.js"><link rel="prefetch" href="/assets/js/15.f247873b.js"><link rel="prefetch" href="/assets/js/150.ead09aca.js"><link rel="prefetch" href="/assets/js/151.971fdf4b.js"><link rel="prefetch" href="/assets/js/152.369c9362.js"><link rel="prefetch" href="/assets/js/153.371edd15.js"><link rel="prefetch" href="/assets/js/155.c68bf602.js"><link rel="prefetch" href="/assets/js/156.304aea8d.js"><link rel="prefetch" href="/assets/js/157.83beef7f.js"><link rel="prefetch" href="/assets/js/158.bb1794b0.js"><link rel="prefetch" href="/assets/js/159.2d54e792.js"><link rel="prefetch" href="/assets/js/16.04336c71.js"><link rel="prefetch" href="/assets/js/160.99d56586.js"><link rel="prefetch" href="/assets/js/161.edf660aa.js"><link rel="prefetch" href="/assets/js/162.0b84606e.js"><link rel="prefetch" href="/assets/js/163.b59e0d60.js"><link rel="prefetch" href="/assets/js/164.d9eb8228.js"><link rel="prefetch" href="/assets/js/165.ca624c79.js"><link rel="prefetch" href="/assets/js/166.025b2ba1.js"><link rel="prefetch" href="/assets/js/167.abc982cc.js"><link rel="prefetch" href="/assets/js/168.27ca13dc.js"><link rel="prefetch" href="/assets/js/169.41e753a2.js"><link rel="prefetch" href="/assets/js/17.43b3c1c8.js"><link rel="prefetch" href="/assets/js/170.626319e1.js"><link rel="prefetch" href="/assets/js/171.a221dddf.js"><link rel="prefetch" href="/assets/js/172.464b2361.js"><link rel="prefetch" href="/assets/js/173.96a3afee.js"><link rel="prefetch" href="/assets/js/174.116607c2.js"><link rel="prefetch" href="/assets/js/175.ea3e8659.js"><link rel="prefetch" href="/assets/js/176.7d7b8afc.js"><link rel="prefetch" href="/assets/js/177.a6e00aa0.js"><link rel="prefetch" href="/assets/js/178.1f93afaf.js"><link rel="prefetch" href="/assets/js/179.3aa00dcd.js"><link rel="prefetch" href="/assets/js/18.d81b44d5.js"><link rel="prefetch" href="/assets/js/180.f8b2b75a.js"><link rel="prefetch" href="/assets/js/181.8e11258a.js"><link rel="prefetch" href="/assets/js/182.22243941.js"><link rel="prefetch" href="/assets/js/183.d051fdf6.js"><link rel="prefetch" href="/assets/js/184.a994075e.js"><link rel="prefetch" href="/assets/js/185.776c7e16.js"><link rel="prefetch" href="/assets/js/186.f1887955.js"><link rel="prefetch" href="/assets/js/187.da0d3626.js"><link rel="prefetch" href="/assets/js/188.8dfc358f.js"><link rel="prefetch" href="/assets/js/189.dcac5a59.js"><link rel="prefetch" href="/assets/js/19.1b3d66e1.js"><link rel="prefetch" href="/assets/js/190.c7e413d0.js"><link rel="prefetch" href="/assets/js/191.d9806121.js"><link rel="prefetch" href="/assets/js/192.869791da.js"><link rel="prefetch" href="/assets/js/193.2d74c4c8.js"><link rel="prefetch" href="/assets/js/194.c73a1909.js"><link rel="prefetch" href="/assets/js/195.e8c74834.js"><link rel="prefetch" href="/assets/js/20.bd5949ec.js"><link rel="prefetch" href="/assets/js/21.3fcf98cf.js"><link rel="prefetch" href="/assets/js/22.2fa1e2e8.js"><link rel="prefetch" href="/assets/js/23.1ae64bb4.js"><link rel="prefetch" href="/assets/js/24.7bdf7387.js"><link rel="prefetch" href="/assets/js/25.392c436e.js"><link rel="prefetch" href="/assets/js/26.58acbd4b.js"><link rel="prefetch" href="/assets/js/27.c725bdd5.js"><link rel="prefetch" href="/assets/js/28.6c9bda1e.js"><link rel="prefetch" href="/assets/js/29.e656b537.js"><link rel="prefetch" href="/assets/js/30.2c326fc7.js"><link rel="prefetch" href="/assets/js/31.e6c9fa30.js"><link rel="prefetch" href="/assets/js/32.c9c88437.js"><link rel="prefetch" href="/assets/js/33.0c53373c.js"><link rel="prefetch" href="/assets/js/34.9821e543.js"><link rel="prefetch" href="/assets/js/35.de8253eb.js"><link rel="prefetch" href="/assets/js/36.d182f929.js"><link rel="prefetch" href="/assets/js/37.9fa79014.js"><link rel="prefetch" href="/assets/js/38.9bebff76.js"><link rel="prefetch" href="/assets/js/39.19a3a2d4.js"><link rel="prefetch" href="/assets/js/4.564edb9d.js"><link rel="prefetch" href="/assets/js/40.cca6955f.js"><link rel="prefetch" href="/assets/js/41.854cd09a.js"><link rel="prefetch" href="/assets/js/42.ca7b612f.js"><link rel="prefetch" href="/assets/js/43.87027d58.js"><link rel="prefetch" href="/assets/js/44.8c2b4f4b.js"><link rel="prefetch" href="/assets/js/45.dffb4e08.js"><link rel="prefetch" href="/assets/js/46.f58049a5.js"><link rel="prefetch" href="/assets/js/47.6854070c.js"><link rel="prefetch" href="/assets/js/48.6cd9fa3d.js"><link rel="prefetch" href="/assets/js/49.e8861afa.js"><link rel="prefetch" href="/assets/js/5.5c31d62f.js"><link rel="prefetch" href="/assets/js/50.703bffab.js"><link rel="prefetch" href="/assets/js/51.6655c373.js"><link rel="prefetch" href="/assets/js/52.deb2eb09.js"><link rel="prefetch" href="/assets/js/53.6e0ed77d.js"><link rel="prefetch" href="/assets/js/54.b05c58ad.js"><link rel="prefetch" href="/assets/js/55.49c8164e.js"><link rel="prefetch" href="/assets/js/56.a5574e6b.js"><link rel="prefetch" href="/assets/js/57.58cb0de4.js"><link rel="prefetch" href="/assets/js/58.52345112.js"><link rel="prefetch" href="/assets/js/59.663ce78d.js"><link rel="prefetch" href="/assets/js/6.a9df34ee.js"><link rel="prefetch" href="/assets/js/60.f06adde2.js"><link rel="prefetch" href="/assets/js/61.170255a1.js"><link rel="prefetch" href="/assets/js/62.9d120050.js"><link rel="prefetch" href="/assets/js/63.70cced6b.js"><link rel="prefetch" href="/assets/js/64.577f3548.js"><link rel="prefetch" href="/assets/js/65.c037b29d.js"><link rel="prefetch" href="/assets/js/66.7dd1045f.js"><link rel="prefetch" href="/assets/js/67.d3aa4d6c.js"><link rel="prefetch" href="/assets/js/68.526dbb61.js"><link rel="prefetch" href="/assets/js/69.58269266.js"><link rel="prefetch" href="/assets/js/7.6609d4d6.js"><link rel="prefetch" href="/assets/js/70.64108f1b.js"><link rel="prefetch" href="/assets/js/71.1e95e0a6.js"><link rel="prefetch" href="/assets/js/72.42e7ec94.js"><link rel="prefetch" href="/assets/js/73.dad4e1c5.js"><link rel="prefetch" href="/assets/js/74.28ea286a.js"><link rel="prefetch" href="/assets/js/75.dd6d4c6f.js"><link rel="prefetch" href="/assets/js/76.ca6539df.js"><link rel="prefetch" href="/assets/js/77.feb13b0e.js"><link rel="prefetch" href="/assets/js/78.321e90e6.js"><link rel="prefetch" href="/assets/js/79.68eb8fcf.js"><link rel="prefetch" href="/assets/js/8.396d51fd.js"><link rel="prefetch" href="/assets/js/80.4edb5321.js"><link rel="prefetch" href="/assets/js/81.735d7e57.js"><link rel="prefetch" href="/assets/js/82.fa120bdf.js"><link rel="prefetch" href="/assets/js/83.bf755f94.js"><link rel="prefetch" href="/assets/js/84.9b32070c.js"><link rel="prefetch" href="/assets/js/85.592aca7c.js"><link rel="prefetch" href="/assets/js/86.4dcd9e73.js"><link rel="prefetch" href="/assets/js/87.a9e546aa.js"><link rel="prefetch" href="/assets/js/88.2a423212.js"><link rel="prefetch" href="/assets/js/89.5f455115.js"><link rel="prefetch" href="/assets/js/9.adb074c6.js"><link rel="prefetch" href="/assets/js/90.5202da0a.js"><link rel="prefetch" href="/assets/js/91.02cee99d.js"><link rel="prefetch" href="/assets/js/92.f16bad0b.js"><link rel="prefetch" href="/assets/js/93.f933634f.js"><link rel="prefetch" href="/assets/js/94.8e7b1d65.js"><link rel="prefetch" href="/assets/js/95.ee0e4a0a.js"><link rel="prefetch" href="/assets/js/96.e21d78c2.js"><link rel="prefetch" href="/assets/js/97.c87e514e.js"><link rel="prefetch" href="/assets/js/98.d123ac92.js"><link rel="prefetch" href="/assets/js/99.92d1b416.js">
    <link rel="stylesheet" href="/assets/css/0.styles.91f57736.css">
  </head>
  <body>
    <div id="app" data-server-rendered="true"><div class="theme-container no-sidebar" data-v-3ba18f14><div data-v-3ba18f14><div id="loader-wrapper" class="loading-wrapper" data-v-041fef5b data-v-3ba18f14 data-v-3ba18f14><div class="loader-main" data-v-041fef5b><div data-v-041fef5b></div><div data-v-041fef5b></div><div data-v-041fef5b></div><div data-v-041fef5b></div></div> <!----> <!----></div> <div class="password-shadow password-wrapper-out" style="display:none;" data-v-68139a52 data-v-3ba18f14 data-v-3ba18f14><h3 class="title" style="display:none;" data-v-68139a52 data-v-68139a52>JavaKeeper</h3> <!----> <label id="box" class="inputBox" style="display:none;" data-v-68139a52 data-v-68139a52><input type="password" value="" data-v-68139a52> <span data-v-68139a52>Konck! Knock!</span> <button data-v-68139a52>OK</button></label> <div class="footer" style="display:none;" data-v-68139a52 data-v-68139a52><span data-v-68139a52><i class="iconfont reco-theme" data-v-68139a52></i> <a target="blank" href="https://vuepress-theme-reco.recoluan.com" data-v-68139a52>vuePress-theme-reco</a></span> <span data-v-68139a52><i class="iconfont reco-copyright" data-v-68139a52></i> <a data-v-68139a52><span data-v-68139a52>海星</span>
            
          <!---->
          2020
        </a></span></div></div> <div class="hide" data-v-3ba18f14><header class="navbar" data-v-3ba18f14><div class="sidebar-button"><svg xmlns="http://www.w3.org/2000/svg" aria-hidden="true" role="img" viewBox="0 0 448 512" class="icon"><path fill="currentColor" d="M436 124H12c-6.627 0-12-5.373-12-12V80c0-6.627 5.373-12 12-12h424c6.627 0 12 5.373 12 12v32c0 6.627-5.373 12-12 12zm0 160H12c-6.627 0-12-5.373-12-12v-32c0-6.627 5.373-12 12-12h424c6.627 0 12 5.373 12 12v32c0 6.627-5.373 12-12 12zm0 160H12c-6.627 0-12-5.373-12-12v-32c0-6.627 5.373-12 12-12h424c6.627 0 12 5.373 12 12v32c0 6.627-5.373 12-12 12z"></path></svg></div> <a href="/" class="home-link router-link-active"><!----> <span class="site-name">JavaKeeper</span></a> <div class="links"><div class="color-picker"><a class="color-button"><i class="iconfont reco-color"></i></a> <div class="color-picker-menu" style="display:none;"><div class="mode-options"><h4 class="title">Choose mode</h4> <ul class="color-mode-options"><li class="dark">dark</li><li class="auto active">auto</li><li class="light">light</li></ul></div></div></div> <div class="search-box"><i class="iconfont reco-search"></i> <input aria-label="Search" autocomplete="off" spellcheck="false" value=""> <!----></div> <nav class="nav-links can-hide"><div class="nav-item"><a href="/java/" class="nav-link"><i class="iconfont undefined"></i>
  Java
</a></div><div class="nav-item"><a href="/data-structure-algorithms/" class="nav-link"><i class="iconfont undefined"></i>
  数据结构与算法
</a></div><div class="nav-item"><a href="/data-store/" class="nav-link"><i class="iconfont undefined"></i>
  数据存储与缓存
</a></div><div class="nav-item"><a href="/interview/" class="nav-link"><i class="iconfont undefined"></i>
  直击面试
</a></div> <a href="https://github.com/Jstarfish/JavaKeeper" target="_blank" rel="noopener noreferrer" class="repo-link"><i class="iconfont reco-github"></i>
    GitHub
    <svg xmlns="http://www.w3.org/2000/svg" aria-hidden="true" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15" class="icon outbound"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path> <polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg></a></nav></div></header> <div class="sidebar-mask" data-v-3ba18f14></div> <aside class="sidebar" data-v-3ba18f14><div class="personal-info-wrapper" data-v-5f6acefd data-v-3ba18f14><!----> <h3 class="name" data-v-5f6acefd>
    海星
  </h3> <div class="num" data-v-5f6acefd><div data-v-5f6acefd><h3 data-v-5f6acefd>0</h3> <h6 data-v-5f6acefd>Article</h6></div> <div data-v-5f6acefd><h3 data-v-5f6acefd>0</h3> <h6 data-v-5f6acefd>Tag</h6></div></div> <hr data-v-5f6acefd></div> <nav class="nav-links"><div class="nav-item"><a href="/java/" class="nav-link"><i class="iconfont undefined"></i>
  Java
</a></div><div class="nav-item"><a href="/data-structure-algorithms/" class="nav-link"><i class="iconfont undefined"></i>
  数据结构与算法
</a></div><div class="nav-item"><a href="/data-store/" class="nav-link"><i class="iconfont undefined"></i>
  数据存储与缓存
</a></div><div class="nav-item"><a href="/interview/" class="nav-link"><i class="iconfont undefined"></i>
  直击面试
</a></div> <a href="https://github.com/Jstarfish/JavaKeeper" target="_blank" rel="noopener noreferrer" class="repo-link"><i class="iconfont reco-github"></i>
    GitHub
    <svg xmlns="http://www.w3.org/2000/svg" aria-hidden="true" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15" class="icon outbound"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path> <polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg></a></nav> <!----> </aside> <div class="password-shadow password-wrapper-in" style="display:none;" data-v-68139a52 data-v-3ba18f14><h3 class="title" style="display:none;" data-v-68139a52 data-v-68139a52></h3> <!----> <label id="box" class="inputBox" style="display:none;" data-v-68139a52 data-v-68139a52><input type="password" value="" data-v-68139a52> <span data-v-68139a52>Konck! Knock!</span> <button data-v-68139a52>OK</button></label> <div class="footer" style="display:none;" data-v-68139a52 data-v-68139a52><span data-v-68139a52><i class="iconfont reco-theme" data-v-68139a52></i> <a target="blank" href="https://vuepress-theme-reco.recoluan.com" data-v-68139a52>vuePress-theme-reco</a></span> <span data-v-68139a52><i class="iconfont reco-copyright" data-v-68139a52></i> <a data-v-68139a52><span data-v-68139a52>海星</span>
            
          <!---->
          2020
        </a></span></div></div> <div data-v-3ba18f14><main class="page"><div class="page-title" style="display:none;"><h1 class="title">KafkaProducer使用介绍、参数配置以及核心源码解析</h1> <div data-v-5d8dbdb4><i class="iconfont reco-account" data-v-5d8dbdb4><span data-v-5d8dbdb4>海星</span></i> <!----> <!----> <!----></div></div> <div class="theme-reco-content content__default" style="display:none;"><h1 id="kafkaproducer使用介绍、参数配置以及核心源码解析"><a href="#kafkaproducer使用介绍、参数配置以及核心源码解析" class="header-anchor">#</a> KafkaProducer使用介绍、参数配置以及核心源码解析</h1> <h2 id="前言"><a href="#前言" class="header-anchor">#</a> 前言</h2> <p>Kafka，作为目前在大数据领域应用最为广泛的消息队列，其内部实现和设计有很多值得深入研究和分析的地方，使用 kafka 首先需要接触到 producer 的开发，然后是 consumer 开发，自 0.8.2.x 版本以后，kafka 提供了 java 版本的 producer 以代替以前 scala 版本的 producer，下面进行 producer 的解析（新版本指的是新版本producer 指的是 kafka-client 包的 <code>org.apache.kafka.clients.producer</code>，而不是 kafka 包下的<code>kafka.producer.Producer</code>）。</p> <h2 id="producer-概要设计"><a href="#producer-概要设计" class="header-anchor">#</a> Producer 概要设计</h2> <p>发送简略流程图</p> <p><img src="https://images2015.cnblogs.com/blog/735367/201702/735367-20170204143450245-662897432.png" alt="img"></p> <p>大体上来说，用户首先构建待发送的消息对象ProducerRecord，然后调用KafkaProducer#send方法进行发送。KafkaProducer接收到消息后首先对其进行序列化，然后通过分区器（partitioner）确定该数据需要发送的 Topic 的分区，kafka 提供了一个默认的分区器，如果消息指定了 key，那么 partitioner 会根据 key 的hash 值来确定目标分区，如果没有指定 key，那么将使用轮询的方式确定目标分区，这样可以最大程度的均衡每个分区的消息，确定分区之后，将会进一步确认该分区的 leader 节点(处理该分区消息读写的主节点)，，最后追加写入到内存中的消息缓冲池(accumulator)。此时KafkaProducer#send方法成功返回。</p> <p>KafkaProducer中还有一个专门的Sender IO线程负责将缓冲池中的消息分批次发送给对应的broker，完成真正的消息发送逻辑。</p> <h2 id="producer-程序开发"><a href="#producer-程序开发" class="header-anchor">#</a> Producer 程序开发</h2> <p>代码很简单，拢共分为 4 步</p> <p><img src="https://i04piccdn.sogoucdn.com/4195c4ece7632219" alt=""></p> <ol><li>配置 producer 参数</li> <li>构造 ProducerRecord 消息</li> <li>调用 send 方法进行发送</li> <li>最后关闭 producer 资源</li></ol> <h3 id="异步提交"><a href="#异步提交" class="header-anchor">#</a> 异步提交</h3> <div class="language-java extra-class"><pre class="language-java"><code><span class="token keyword">public</span> <span class="token keyword">static</span> <span class="token keyword">void</span> <span class="token function">main</span><span class="token punctuation">(</span><span class="token class-name">String</span><span class="token punctuation">[</span><span class="token punctuation">]</span> args<span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token class-name">Properties</span> properties <span class="token operator">=</span> <span class="token keyword">new</span> <span class="token class-name">Properties</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token comment">// Kafka 服务端的主机名和端口号</span>
    properties<span class="token punctuation">.</span><span class="token function">put</span><span class="token punctuation">(</span><span class="token string">&quot;bootstrap.servers&quot;</span><span class="token punctuation">,</span> <span class="token string">&quot;10.202.253.240:9092&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token comment">// 等待所有副本节点的应答</span>
    properties<span class="token punctuation">.</span><span class="token function">put</span><span class="token punctuation">(</span><span class="token string">&quot;acks&quot;</span><span class="token punctuation">,</span> <span class="token string">&quot;all&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token comment">// 消息发送最大尝试次数</span>
    properties<span class="token punctuation">.</span><span class="token function">put</span><span class="token punctuation">(</span><span class="token string">&quot;retries&quot;</span><span class="token punctuation">,</span> <span class="token number">0</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token comment">// 一批消息处理大小</span>
    properties<span class="token punctuation">.</span><span class="token function">put</span><span class="token punctuation">(</span><span class="token string">&quot;batch.size&quot;</span><span class="token punctuation">,</span> <span class="token number">16384</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token comment">// 请求延时</span>
    properties<span class="token punctuation">.</span><span class="token function">put</span><span class="token punctuation">(</span><span class="token string">&quot;linger.ms&quot;</span><span class="token punctuation">,</span> <span class="token number">1</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token comment">// 发送缓存区内存大小</span>
    properties<span class="token punctuation">.</span><span class="token function">put</span><span class="token punctuation">(</span><span class="token string">&quot;buffer.memory&quot;</span><span class="token punctuation">,</span> <span class="token number">33554432</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token comment">// key 序列化</span>
  properties<span class="token punctuation">.</span><span class="token function">put</span><span class="token punctuation">(</span><span class="token string">&quot;key.serializer&quot;</span><span class="token punctuation">,</span><span class="token string">&quot;org.apache.kafka.common.serialization.StringSerializer&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token comment">// value 序列化</span>
  properties<span class="token punctuation">.</span><span class="token function">put</span><span class="token punctuation">(</span><span class="token string">&quot;value.serializer&quot;</span><span class="token punctuation">,</span><span class="token string">&quot;org.apache.kafka.common.serialization.StringSerializer&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token class-name">Producer</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">String</span><span class="token punctuation">,</span> <span class="token class-name">String</span><span class="token punctuation">&gt;</span></span> producer <span class="token operator">=</span> <span class="token keyword">new</span> <span class="token class-name">KafkaProducer</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">String</span><span class="token punctuation">,</span> <span class="token class-name">String</span><span class="token punctuation">&gt;</span></span><span class="token punctuation">(</span>properties<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">for</span> <span class="token punctuation">(</span><span class="token keyword">int</span> i <span class="token operator">=</span> <span class="token number">0</span><span class="token punctuation">;</span> i <span class="token operator">&lt;</span> <span class="token number">10</span><span class="token punctuation">;</span> i<span class="token operator">++</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        producer<span class="token punctuation">.</span><span class="token function">send</span><span class="token punctuation">(</span><span class="token keyword">new</span> <span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">String</span><span class="token punctuation">,</span> <span class="token class-name">String</span><span class="token punctuation">&gt;</span></span><span class="token punctuation">(</span><span class="token string">&quot;test&quot;</span><span class="token punctuation">,</span>
                <span class="token class-name">Integer</span><span class="token punctuation">.</span><span class="token function">toString</span><span class="token punctuation">(</span>i<span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token string">&quot;hello world-&quot;</span> <span class="token operator">+</span> i<span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span>
    producer<span class="token punctuation">.</span><span class="token function">close</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre></div><blockquote><p>如果发送失败，也可能 server.properties 配置的问题，不允许远程访问</p> <ul><li>去掉注释，listeners=PLAINTEXT://:9092</li> <li>把advertised.listeners值改为PLAINTEXT://host_ip:9092</li></ul></blockquote> <h3 id="同步提交"><a href="#同步提交" class="header-anchor">#</a> 同步提交</h3> <p>Kafka 的 Producer 发送消息采用的是异步发送的方式，<code>KafkaProducer</code> 的 <code>send</code> 方法返回 <code>Future</code> 对象，所以我们可以手动调用 <code>Future</code> 对象的 <code>get</code> 方法实现同步：</p> <div class="language-java extra-class"><pre class="language-java"><code>producer<span class="token punctuation">.</span><span class="token function">send</span><span class="token punctuation">(</span><span class="token keyword">new</span> <span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">String</span><span class="token punctuation">,</span> <span class="token class-name">String</span><span class="token punctuation">&gt;</span></span><span class="token punctuation">(</span><span class="token string">&quot;test&quot;</span><span class="token punctuation">,</span>
        <span class="token class-name">Integer</span><span class="token punctuation">.</span><span class="token function">toString</span><span class="token punctuation">(</span>i<span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token string">&quot;hello world-&quot;</span> <span class="token operator">+</span> i<span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">get</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
</code></pre></div><p>get 方法将阻塞，直到返回结果 RecordMetadata，所以可以看成是同步的。</p> <h3 id="异步待回调函数"><a href="#异步待回调函数" class="header-anchor">#</a> 异步待回调函数</h3> <p>上边的异步提交方式，可以称为发送并忘记（不关心消息是否正常到达，对返回结果不做任何判断处理），所以 kafak 提供了一个带回调函数的 send 方法</p> <div class="language-java extra-class"><pre class="language-java"><code><span class="token class-name">Future</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">RecordMetadata</span><span class="token punctuation">&gt;</span></span> <span class="token function">send</span><span class="token punctuation">(</span><span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> producer<span class="token punctuation">,</span> <span class="token class-name">Callback</span> callback<span class="token punctuation">)</span><span class="token punctuation">;</span>
</code></pre></div><p>Callback 是一个回调接口，在消息发送完成之后可以回调我们自定义的实现</p> <div class="language-java extra-class"><pre class="language-java"><code><span class="token keyword">for</span> <span class="token punctuation">(</span><span class="token keyword">int</span> i <span class="token operator">=</span> <span class="token number">0</span><span class="token punctuation">;</span> i <span class="token operator">&lt;</span> <span class="token number">10</span><span class="token punctuation">;</span> i<span class="token operator">++</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
    producer<span class="token punctuation">.</span><span class="token function">send</span><span class="token punctuation">(</span><span class="token keyword">new</span> <span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">String</span><span class="token punctuation">,</span> <span class="token class-name">String</span><span class="token punctuation">&gt;</span></span><span class="token punctuation">(</span><span class="token string">&quot;test&quot;</span><span class="token punctuation">,</span>
            <span class="token class-name">Integer</span><span class="token punctuation">.</span><span class="token function">toString</span><span class="token punctuation">(</span>i<span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token string">&quot;hello world-&quot;</span> <span class="token operator">+</span> i<span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token keyword">new</span> <span class="token class-name">Callback</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token annotation punctuation">@Override</span>
        <span class="token keyword">public</span> <span class="token keyword">void</span> <span class="token function">onCompletion</span><span class="token punctuation">(</span><span class="token class-name">RecordMetadata</span> recordMetadata<span class="token punctuation">,</span> <span class="token class-name">Exception</span> e<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">if</span> <span class="token punctuation">(</span>e <span class="token operator">==</span> <span class="token keyword">null</span><span class="token punctuation">)</span><span class="token punctuation">{</span>
                <span class="token class-name">System</span><span class="token punctuation">.</span>out<span class="token punctuation">.</span><span class="token function">println</span><span class="token punctuation">(</span><span class="token string">&quot;TopicName : &quot;</span> <span class="token operator">+</span> recordMetadata<span class="token punctuation">.</span><span class="token function">topic</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">+</span> <span class="token string">&quot; Partiton : &quot;</span> <span class="token operator">+</span> recordMetadata
                        <span class="token punctuation">.</span><span class="token function">partition</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">+</span> <span class="token string">&quot; Offset : &quot;</span> <span class="token operator">+</span> recordMetadata<span class="token punctuation">.</span><span class="token function">offset</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span>
            <span class="token keyword">else</span> <span class="token punctuation">{</span>
                <span class="token comment">//进行异常处理</span>
            <span class="token punctuation">}</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre></div><p>同样的也能获取结果，回调的时候会传递两个参数：</p> <ul><li><code>RecordMetadata</code> 和上文一致的消息发送成功后的元数据。</li> <li><code>Exception</code> 消息发送过程中的异常信息。</li></ul> <p>但是这两个参数并不会同时都有数据，只有发送失败才会有异常信息，同时发送元数据为空。</p> <p><strong>同步发送消息</strong></p> <ul><li><p>优点：可以保证每条消息准确无误的写入了 broker，对于立即需要发送结果的情况非常适用，在 producer 故障或者宕机的情况也可以保证结果的正确性</p></li> <li><p>缺点：由于同步发送需要每条消息都需要及时发送到 broker，没有缓冲批量操作，性能较低</p></li></ul> <p><strong>异步发送消息</strong></p> <ul><li><p>优点：可以通过缓冲池对消息进行缓冲，然后进行消息的批量发送，大量减少了和 broker 的交互频率，性能极高，可以通过回调机制获取发送结果</p></li> <li><p>缺点：在 producer 直接断电或者重启等故障，将有可能丢失消息发送结果，对消息准确性要求很高的场景不适用</p></li></ul> <h3 id="带拦截器的-producer"><a href="#带拦截器的-producer" class="header-anchor">#</a> 带拦截器的 Producer</h3> <p>TODO</p> <h2 id="producer参数说明"><a href="#producer参数说明" class="header-anchor">#</a> Producer参数说明</h2> <ul><li><p><strong>bootstrap.servers</strong>：Kafka 集群信息列表，用于建立到 Kafka 集群的初始连接，如果集群中机器数很多，只需要指定部分的机器主机的信息即可，不管指定多少台，producer 都会通过其中一台机器发现集群中所有的 broker，格式为 <code>hostname1:port,hostname2:port,hostname3:port</code></p></li> <li><p><strong>key.serializer</strong>：任何消息发送到 broker 格式都是字节数组，因此在发送到 broker 之前需要进行序列化，该参数用于对 ProducerRecord 中的 key进行序列化</p></li> <li><p><strong>value.serializer</strong>：该参数用于对 ProducerRecord 中的 value 进行序列化</p> <blockquote><p>kafka 默认提供了常用的序列化类，也可以通过实现 <code>org.apache.kafka.common.serialization.Serializer</code> 实现定义的序列化，Kafka 提供的序列化类如下：</p> <p><img src="https://static01.imgkr.com/temp/0997ff0db5a04501bc3438ed3b7402c9.png" alt=""></p></blockquote></li> <li><p>**acks：**ack 具有3个取值 0、1 和 -1(all)</p> <ul><li><p>acks=0：producer 不等待 broker 的 ack，这一操作提供了一个最低的延迟，broker 一接收到还没有写入磁盘就已经返回，吞吐量最高，当 broker 故障时有可能<strong>丢失数据</strong>；</p></li> <li><p>ack=1：producer 等待 broker 的 ack，partition 的 leader 落盘成功后返回 ack，如果在 follower 同步成功之前 leader 故障，那么将会<strong>丢失数据</strong></p></li> <li><p>acks=-1(all)：producer 等待 broker 的 ack，partition 的 leader 和 follower 全部落盘成功后才返回 ack。但是如果在 follower 同步完成后，broker 发送 ack 之前，leader 发生故障，那么就会造成<strong>数据重复</strong></p></li></ul></li> <li><p><strong>buffer.memory</strong>：该参数用于设置发送缓冲池的内存大小，单位是字节。默认值 33554432KB（32M）</p></li> <li><p><strong>max.block.ms</strong>：当 producer 缓冲满了之后，阻塞的时间</p></li> <li><p><strong>compression.type</strong>：压缩类型，目前 kafka 支持四种种压缩类型 <code>gzip</code>、<code>snappy</code>、<code>lz4</code>、<code>zstd</code>，性能依次递增。默认值none（不压缩）</p></li> <li><p><strong>retries</strong>：重试次数，0 表示不进行重试。默认值 2147483647</p></li> <li><p><strong>batch.size</strong>：批次处理大小，通常增加该参数，可以提升 producer 的吞吐量，默认值 16384</p></li> <li><p><strong>linger.ms</strong>：发送时延，和 batch.size 任满足其一条件，就会进行发送，减少网络IO，节省带宽之用。原理就是把原本需要多次发送的小batch，通过引入延时的方式合并成大batch发送，减少了网络传输的压力，从而提升吞吐量。当然，也会引入延时</p></li> <li><p><strong>max.request.size</strong>：控制 producer 发送请求的大小</p></li> <li><p><strong>receive.buffer.bytes</strong>：读取数据时要使用的 TCP 接收缓冲区（SO_RCVBUF）的大小。如果值为 -1，则将使用OS默认值。默认值 32768</p></li> <li><p><strong>send. buffer.bytes</strong>：发送数据时使用的 TCP 发送缓冲区（SO_SNDBUF）的大小。如果值为-1，则将使用OS默认值。默认值131072</p></li> <li><p><strong>request.timeout.ms</strong>：生产者发送数据时等待服务器返回响应的时间。默认值 30000ms</p></li></ul> <h2 id="消息分区机制"><a href="#消息分区机制" class="header-anchor">#</a> 消息分区机制</h2> <p>消息发送时都被发送到一个 topic，其本质就是一个目录，而 topic 是由一些 Partition Logs(分区日志)组成</p> <p><strong>分区的原因：</strong></p> <ol><li><strong>方便在集群中扩展</strong>，每个 Partition 可以通过调整以适应它所在的机器，而一个 topic 又可以有多个 Partition 组成，因此整个集群就可以适应任意大小的数据了；</li> <li><strong>可以提高并发</strong>，因为可以以 Partition 为单位读写了。</li></ol> <p><strong>分区的原则：</strong></p> <p>我们需要将 producer 发送的数据封装成一个 ProducerRecord 对象。</p> <div class="language-java extra-class"><pre class="language-java"><code><span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Integer</span> partition<span class="token punctuation">,</span> <span class="token class-name">Long</span> timestamp<span class="token punctuation">,</span> <span class="token class-name">K</span> key<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">,</span> <span class="token class-name">Iterable</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">Header</span><span class="token punctuation">&gt;</span></span> headers<span class="token punctuation">)</span>
<span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Integer</span> partition<span class="token punctuation">,</span> <span class="token class-name">Long</span> timestamp<span class="token punctuation">,</span> <span class="token class-name">K</span> key<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">)</span>
<span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Integer</span> partition<span class="token punctuation">,</span> <span class="token class-name">K</span> key<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">,</span> <span class="token class-name">Iterable</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">Header</span><span class="token punctuation">&gt;</span></span> headers<span class="token punctuation">)</span>
<span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Integer</span> partition<span class="token punctuation">,</span> <span class="token class-name">K</span> key<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">)</span>
<span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">K</span> key<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">)</span>
<span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">)</span><span class="token class-name">Copy</span> <span class="token keyword">to</span> <span class="token namespace">clipboardErrorCopied</span>
</code></pre></div><ol><li>指明 partition 的情况下，直接将指明的值直接作为 partiton 值；</li> <li>没有指明 partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值；</li> <li>既没有 partition 值又没有 key 值的情况下，第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition 值，也就是常说的 round-robin 算法。</li></ol> <p><strong>自定义分区器</strong></p> <p>可以通过实现 <code>org.apache.kafka.clients.producer.Partitioner</code> 自定分区策略，在构造 KafkaProducer 时配置参数 <code>partitioner.class</code> 为自定义的分区类即可</p> <div class="language-java extra-class"><pre class="language-java"><code><span class="token keyword">public</span> <span class="token keyword">class</span> <span class="token class-name">MyPartitioner</span> <span class="token keyword">implements</span> <span class="token class-name">Partitioner</span> <span class="token punctuation">{</span>

    <span class="token annotation punctuation">@Override</span>
    <span class="token keyword">public</span> <span class="token keyword">int</span> <span class="token function">partition</span><span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Object</span> key<span class="token punctuation">,</span> <span class="token keyword">byte</span><span class="token punctuation">[</span><span class="token punctuation">]</span> keyBytes<span class="token punctuation">,</span> <span class="token class-name">Object</span> value<span class="token punctuation">,</span>
                         <span class="token keyword">byte</span><span class="token punctuation">[</span><span class="token punctuation">]</span> valueBytes<span class="token punctuation">,</span> <span class="token class-name">Cluster</span> cluster<span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token keyword">return</span> <span class="token number">0</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span>

    <span class="token annotation punctuation">@Override</span>
    <span class="token keyword">public</span> <span class="token keyword">void</span> <span class="token function">close</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>

    <span class="token punctuation">}</span>

    <span class="token annotation punctuation">@Override</span>
    <span class="token keyword">public</span> <span class="token keyword">void</span> <span class="token function">configure</span><span class="token punctuation">(</span><span class="token class-name">Map</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">String</span><span class="token punctuation">,</span> <span class="token operator">?</span><span class="token punctuation">&gt;</span></span> configs<span class="token punctuation">)</span> <span class="token punctuation">{</span>

    <span class="token punctuation">}</span>
<span class="token punctuation">}</span>
</code></pre></div><h2 id="内部原理"><a href="#内部原理" class="header-anchor">#</a> 内部原理</h2> <p>Java producer(区别于Scala producer)是双线程的设计，分为KafkaProducer用户主线程和Sender线程</p> <p>producer 总共创建两个线程：执行KafkaPrducer#send逻辑的线程——我们称之为“用户主线程”；执行发送逻辑的IO线程——我们称之为“Sender线程”</p> <h3 id="_1-序列化-计算目标分区"><a href="#_1-序列化-计算目标分区" class="header-anchor">#</a> 1. 序列化+计算目标分区</h3> <p>这是KafkaProducer#send逻辑的第一步，即为待发送消息进行序列化并计算目标分区，如下图所示：</p> <p><img src="https://images2015.cnblogs.com/blog/735367/201702/735367-20170204162034104-1142811823.png" alt="img"></p> <p>如上图所示，一条所属topic是&quot;test&quot;，消息体是&quot;message&quot;的消息被序列化之后结合KafkaProducer缓存的元数据(比如该topic分区数信息等)共同传给后面的Partitioner实现类进行目标分区的计算。</p> <h3 id="_2-追加写入消息缓冲区-accumulator"><a href="#_2-追加写入消息缓冲区-accumulator" class="header-anchor">#</a> 2. 追加写入消息缓冲区(accumulator)</h3> <p>producer 创建时会创建一个默认32MB(由buffer.memory参数指定)的accumulator缓冲区，专门保存待发送的消息。除了之前在“关键参数”段落中提到的linger.ms和batch.size等参数之外，该数据结构中还包含了一个特别重要的集合信息：消息批次信息(batches)。该集合本质上是一个HashMap，里面分别保存了每个topic分区下的batch队列，即前面说的批次是按照topic分区进行分组的。这样发往不同分区的消息保存在对应分区下的batch队列中。举个简单的例子，假设消息M1, M2被发送到test的0分区但属于不同的batch，M3分送到test的1分区，那么batches中包含的信息就是：{&quot;test-0&quot; -&gt; [batch1, batch2], &quot;test-1&quot; -&gt; [batch3]}</p> <p>单个topic分区下的batch队列中保存的是若干个消息批次。每个batch中最重要的3个组件包括：</p> <ul><li>compressor: 负责执行追加写入操作</li> <li>batch缓冲区：由batch.size参数控制，消息被真正追加写入到的地方</li> <li>thunks：保存消息回调逻辑的集合</li></ul> <p>这一步的目的就是将待发送的消息写入消息缓冲池中，具体流程如下图所示：</p> <p><img src="https://images2015.cnblogs.com/blog/735367/201702/735367-20170204164910854-2033381282.png" alt=""></p> <p>okay！这一步执行完毕之后理论上讲KafkaProducer.send方法就执行完毕了，用户主线程所做的事情就是等待Sender线程发送消息并执行返回结果了。</p> <h3 id="_3-sender线程预处理及消息发送"><a href="#_3-sender线程预处理及消息发送" class="header-anchor">#</a> 3. Sender线程预处理及消息发送</h3> <p>此时，该Sender线程登场了。严格来说，Sender线程自KafkaProducer创建后就一直都在运行着 。它的工作流程基本上是这样的：</p> <ol><li>不断轮询缓冲区寻找已做好发送准备的分区</li> <li>将轮询获得的各个batch按照目标分区所在的leader broker进行分组</li> <li>将分组后的batch通过底层创建的Socket连接发送给各个broker</li> <li>等待服务器端发送response回来</li></ol> <p>为了说明上的方便，我还是基于图的方式来解释Sender线程的工作原理：</p> <p><img src="https://images2015.cnblogs.com/blog/735367/201702/735367-20170204164759886-242426092.png" alt="img"></p> <h3 id="_4-sender线程处理response"><a href="#_4-sender线程处理response" class="header-anchor">#</a> 4. Sender线程处理response</h3> <p>上图中Sender线程会发送PRODUCE请求给对应的broker，broker处理完毕之后发送对应的PRODUCE response。一旦Sender线程接收到response将依次(按照消息发送顺序)调用batch中的回调方法，如下图所示：</p> <p><img src="https://images2015.cnblogs.com/blog/735367/201702/735367-20170204170043386-56898873.png" alt="img"></p> <p>做完这一步，producer发送消息就可以算作是100%完成了。通过这4步我们可以看到新版本producer发送事件完全是异步过程。因此在调优producer前我们就需要搞清楚性能瓶颈到底是在用户主线程还是在Sender线程。具体的性能测试方法以及调优方法以后有机会的话我再写一篇出来和大家讨论。</p> <h2 id="源码解毒"><a href="#源码解毒" class="header-anchor">#</a> 源码解毒</h2> <h3 id="producer-send"><a href="#producer-send" class="header-anchor">#</a> producer send()</h3> <div class="language-java extra-class"><pre class="language-java"><code><span class="token comment">// 1.异步向一个 topic 发送数据</span>
<span class="token keyword">public</span> <span class="token class-name">Future</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">RecordMetadata</span><span class="token punctuation">&gt;</span></span> <span class="token function">send</span><span class="token punctuation">(</span><span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> <span class="token keyword">record</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token keyword">return</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">send</span><span class="token punctuation">(</span><span class="token keyword">record</span><span class="token punctuation">,</span> <span class="token punctuation">(</span><span class="token class-name">Callback</span><span class="token punctuation">)</span><span class="token keyword">null</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>


<span class="token comment">// 异步向一个 topic 发送数据，并注册回调函数，在回调函数中接受发送响应</span>
<span class="token keyword">public</span> <span class="token class-name">Future</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">RecordMetadata</span><span class="token punctuation">&gt;</span></span> <span class="token function">send</span><span class="token punctuation">(</span><span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> <span class="token keyword">record</span><span class="token punctuation">,</span> <span class="token class-name">Callback</span> callback<span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token comment">//如果有自定义拦截器，会去处理拦截器的Producer</span>
    <span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> interceptedRecord <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>interceptors<span class="token punctuation">.</span><span class="token function">onSend</span><span class="token punctuation">(</span><span class="token keyword">record</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">return</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">doSend</span><span class="token punctuation">(</span>interceptedRecord<span class="token punctuation">,</span> callback<span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre></div><p><code>send()</code> 方法通过重载实现带回调和不带回调的参数，最终都调用 <strong>doSend()</strong> 方法</p> <div class="language-java extra-class"><pre class="language-java"><code><span class="token keyword">private</span> <span class="token class-name">Future</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">RecordMetadata</span><span class="token punctuation">&gt;</span></span> <span class="token function">doSend</span><span class="token punctuation">(</span><span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> <span class="token keyword">record</span><span class="token punctuation">,</span> <span class="token class-name">Callback</span> callback<span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token class-name">TopicPartition</span> tp <span class="token operator">=</span> <span class="token keyword">null</span><span class="token punctuation">;</span>

    <span class="token keyword">try</span> <span class="token punctuation">{</span>
        <span class="token comment">//1、检查producer实例是否已关闭,如果关闭则抛出异常</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">throwIfProducerClosed</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>

        <span class="token class-name">KafkaProducer</span><span class="token punctuation">.</span><span class="token class-name">ClusterAndWaitTime</span> clusterAndWaitTime<span class="token punctuation">;</span>
        <span class="token keyword">try</span> <span class="token punctuation">{</span>
            <span class="token comment">//2、确保topic的元数据(metadata)是可用的</span>
            clusterAndWaitTime <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">waitOnMetadata</span><span class="token punctuation">(</span><span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">topic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">partition</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>maxBlockTimeMs<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span> <span class="token keyword">catch</span> <span class="token punctuation">(</span><span class="token class-name">KafkaException</span> var19<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>metadata<span class="token punctuation">.</span><span class="token function">isClosed</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">KafkaException</span><span class="token punctuation">(</span><span class="token string">&quot;Producer closed while send in progress&quot;</span><span class="token punctuation">,</span> var19<span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span>

            <span class="token keyword">throw</span> var19<span class="token punctuation">;</span>
        <span class="token punctuation">}</span>

        <span class="token keyword">long</span> remainingWaitMs <span class="token operator">=</span> <span class="token class-name">Math</span><span class="token punctuation">.</span><span class="token function">max</span><span class="token punctuation">(</span><span class="token number">0L</span><span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>maxBlockTimeMs <span class="token operator">-</span> clusterAndWaitTime<span class="token punctuation">.</span>waitedOnMetadataMs<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token class-name">Cluster</span> cluster <span class="token operator">=</span> clusterAndWaitTime<span class="token punctuation">.</span>cluster<span class="token punctuation">;</span>
		<span class="token comment">//序列化key 和 value</span>
        <span class="token keyword">byte</span><span class="token punctuation">[</span><span class="token punctuation">]</span> serializedKey<span class="token punctuation">;</span>
        <span class="token keyword">try</span> <span class="token punctuation">{</span>
            serializedKey <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>keySerializer<span class="token punctuation">.</span><span class="token function">serialize</span><span class="token punctuation">(</span><span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">topic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">headers</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">key</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span> <span class="token keyword">catch</span> <span class="token punctuation">(</span><span class="token class-name">ClassCastException</span> var18<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">SerializationException</span><span class="token punctuation">(</span><span class="token string">&quot;Can't convert key of class &quot;</span> <span class="token operator">+</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">key</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">getClass</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">getName</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">+</span> <span class="token string">&quot; to class &quot;</span> <span class="token operator">+</span> <span class="token keyword">this</span><span class="token punctuation">.</span>producerConfig<span class="token punctuation">.</span><span class="token function">getClass</span><span class="token punctuation">(</span><span class="token string">&quot;key.serializer&quot;</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">getName</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">+</span> <span class="token string">&quot; specified in key.serializer&quot;</span><span class="token punctuation">,</span> var18<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>

        <span class="token keyword">byte</span><span class="token punctuation">[</span><span class="token punctuation">]</span> serializedValue<span class="token punctuation">;</span>
        <span class="token keyword">try</span> <span class="token punctuation">{</span>
            serializedValue <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>valueSerializer<span class="token punctuation">.</span><span class="token function">serialize</span><span class="token punctuation">(</span><span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">topic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">headers</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">value</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span> <span class="token keyword">catch</span> <span class="token punctuation">(</span><span class="token class-name">ClassCastException</span> var17<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">SerializationException</span><span class="token punctuation">(</span><span class="token string">&quot;Can't convert value of class &quot;</span> <span class="token operator">+</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">value</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">getClass</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">getName</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">+</span> <span class="token string">&quot; to class &quot;</span> <span class="token operator">+</span> <span class="token keyword">this</span><span class="token punctuation">.</span>producerConfig<span class="token punctuation">.</span><span class="token function">getClass</span><span class="token punctuation">(</span><span class="token string">&quot;value.serializer&quot;</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">getName</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">+</span> <span class="token string">&quot; specified in value.serializer&quot;</span><span class="token punctuation">,</span> var17<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>

        <span class="token comment">//获取分区信息,如果ProducerRecord指定了分区信息就使用该指定分区否则通过计算获取</span>
        <span class="token keyword">int</span> partition <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">partition</span><span class="token punctuation">(</span><span class="token keyword">record</span><span class="token punctuation">,</span> serializedKey<span class="token punctuation">,</span> serializedValue<span class="token punctuation">,</span> cluster<span class="token punctuation">)</span><span class="token punctuation">;</span>
        tp <span class="token operator">=</span> <span class="token keyword">new</span> <span class="token class-name">TopicPartition</span><span class="token punctuation">(</span><span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">topic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> partition<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">setReadOnly</span><span class="token punctuation">(</span><span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">headers</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token class-name">Header</span><span class="token punctuation">[</span><span class="token punctuation">]</span> headers <span class="token operator">=</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">headers</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">toArray</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token comment">//估算消息的字节大小</span>
        <span class="token keyword">int</span> serializedSize <span class="token operator">=</span> <span class="token class-name">AbstractRecords</span><span class="token punctuation">.</span><span class="token function">estimateSizeInBytesUpperBound</span><span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>apiVersions<span class="token punctuation">.</span><span class="token function">maxUsableProduceMagic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>compressionType<span class="token punctuation">,</span> serializedKey<span class="token punctuation">,</span> serializedValue<span class="token punctuation">,</span> headers<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token comment">//确保消息大小不超过发送请求最大值(max.request.size)或者发送缓冲池发小(buffer.memory)</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">ensureValidRecordSize</span><span class="token punctuation">(</span>serializedSize<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">long</span> timestamp <span class="token operator">=</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">timestamp</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">==</span> <span class="token keyword">null</span> <span class="token operator">?</span> <span class="token keyword">this</span><span class="token punctuation">.</span>time<span class="token punctuation">.</span><span class="token function">milliseconds</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">:</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">timestamp</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">trace</span><span class="token punctuation">(</span><span class="token string">&quot;Sending record {} with callback {} to topic {} partition {}&quot;</span><span class="token punctuation">,</span> <span class="token keyword">new</span> <span class="token class-name">Object</span><span class="token punctuation">[</span><span class="token punctuation">]</span><span class="token punctuation">{</span><span class="token keyword">record</span><span class="token punctuation">,</span> callback<span class="token punctuation">,</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">topic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> partition<span class="token punctuation">}</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token class-name">Callback</span> interceptCallback <span class="token operator">=</span> <span class="token keyword">new</span> <span class="token class-name">KafkaProducer</span><span class="token punctuation">.</span><span class="token class-name">InterceptorCallback</span><span class="token punctuation">(</span>callback<span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>interceptors<span class="token punctuation">,</span> tp<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token comment">//是否使用事务</span>
        <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager <span class="token operator">!=</span> <span class="token keyword">null</span> <span class="token operator">&amp;&amp;</span> <span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">isTransactional</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">maybeAddPartitionToTransaction</span><span class="token punctuation">(</span>tp<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>
		<span class="token comment">//向 accumulator 中追加数据</span>
        <span class="token class-name">RecordAppendResult</span> result <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>accumulator<span class="token punctuation">.</span><span class="token function">append</span><span class="token punctuation">(</span>tp<span class="token punctuation">,</span> timestamp<span class="token punctuation">,</span> serializedKey<span class="token punctuation">,</span> serializedValue<span class="token punctuation">,</span> headers<span class="token punctuation">,</span> interceptCallback<span class="token punctuation">,</span> remainingWaitMs<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token comment">//如果 batch 已经满了,唤醒 sender 线程发送数据</span>
        <span class="token keyword">if</span> <span class="token punctuation">(</span>result<span class="token punctuation">.</span>batchIsFull <span class="token operator">||</span> result<span class="token punctuation">.</span>newBatchCreated<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">trace</span><span class="token punctuation">(</span><span class="token string">&quot;Waking up the sender since topic {} partition {} is either full or getting a new batch&quot;</span><span class="token punctuation">,</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">topic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> partition<span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token keyword">this</span><span class="token punctuation">.</span>sender<span class="token punctuation">.</span><span class="token function">wakeup</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>

        <span class="token keyword">return</span> result<span class="token punctuation">.</span>future<span class="token punctuation">;</span>
    <span class="token punctuation">}</span> <span class="token keyword">catch</span> <span class="token punctuation">(</span><span class="token class-name">ApiException</span> var20<span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span>
    <span class="token punctuation">}</span> 
    <span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span>
<span class="token punctuation">}</span>
</code></pre></div><p>doSend() 方法主要分为10步完成：</p> <ol><li><p>检查producer实例是否已关闭,如果关闭则抛出异常</p></li> <li><p>确保topic的元数据(metadata)是可用的</p></li> <li><p>序列化ProducerRecord中的key</p></li> <li><p>序列化ProducerRecord中的value</p></li> <li><p>确定分区信息,如果构造ProducerRecord指定了分区信息就使用该指定分区否则通过计算获取</p></li> <li><p>估算整条消息的字节大小</p></li> <li><p>确保消息大小不超过发送请求最大值(max.request.size)或者发送缓冲池发小(buffer.memory)，如果超过则抛出异常</p></li> <li><p>是否使用事务，如果使用则按照事务流程进行</p></li> <li><p>向 accumulator 中追加数据</p></li> <li><p>如果 batch 已经满了,唤醒 sender 线程发送数据</p></li></ol> <p><img src="https://img2018.cnblogs.com/blog/1465200/201909/1465200-20190910145036444-1215527333.png" alt=""></p> <h3 id="具体的发送过程"><a href="#具体的发送过程" class="header-anchor">#</a> 具体的发送过程</h3> <h4 id="获取-topic-的-metadata-信息-waitonmetadata"><a href="#获取-topic-的-metadata-信息-waitonmetadata" class="header-anchor">#</a> 获取 topic 的 metadata 信息——waitOnMetadata()</h4> <p>Producer 通过 <code>waitOnMetadata()</code> 方法来获取对应 topic 的 metadata 信息</p> <div class="language-java extra-class"><pre class="language-java"><code><span class="token comment">//如果元数据Topic列表中还没有该Topic，则将其添加到元数据Topic列表中，并重置过期时间</span>
<span class="token keyword">private</span> <span class="token class-name">KafkaProducer</span><span class="token punctuation">.</span><span class="token class-name">ClusterAndWaitTime</span> <span class="token function">waitOnMetadata</span><span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Integer</span> partition<span class="token punctuation">,</span> <span class="token keyword">long</span> maxWaitMs<span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">InterruptedException</span> <span class="token punctuation">{</span>
    <span class="token comment">//首先从元数据中获取集群信息</span>
    <span class="token class-name">Cluster</span> cluster <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>metadata<span class="token punctuation">.</span><span class="token function">fetch</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token comment">//集群的无效topic列表包含该topic那么抛出异常</span>
    <span class="token keyword">if</span> <span class="token punctuation">(</span>cluster<span class="token punctuation">.</span><span class="token function">invalidTopics</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">contains</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">InvalidTopicException</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span> <span class="token keyword">else</span> <span class="token punctuation">{</span>
        <span class="token comment">//否则将topic添加到Topic列表中</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span>metadata<span class="token punctuation">.</span><span class="token function">add</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token comment">//获取该topic的分区数</span>
        <span class="token class-name">Integer</span> partitionsCount <span class="token operator">=</span> cluster<span class="token punctuation">.</span><span class="token function">partitionCountForTopic</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token comment">//如果存在缓存的元数据，并且未指定分区的或者在已知分区范围内，那么返回缓存的元数据</span>
        <span class="token keyword">if</span> <span class="token punctuation">(</span>partitionsCount <span class="token operator">==</span> <span class="token keyword">null</span> <span class="token operator">||</span> partition <span class="token operator">!=</span> <span class="token keyword">null</span> <span class="token operator">&amp;&amp;</span> partition <span class="token operator">&gt;=</span> partitionsCount<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">long</span> begin <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>time<span class="token punctuation">.</span><span class="token function">milliseconds</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token keyword">long</span> remainingWaitMs <span class="token operator">=</span> maxWaitMs<span class="token punctuation">;</span>

            <span class="token keyword">long</span> elapsed<span class="token punctuation">;</span>
            <span class="token keyword">do</span> <span class="token punctuation">{</span>
                <span class="token keyword">if</span> <span class="token punctuation">(</span>partition <span class="token operator">!=</span> <span class="token keyword">null</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                    <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">trace</span><span class="token punctuation">(</span><span class="token string">&quot;Requesting metadata update for partition {} of topic {}.&quot;</span><span class="token punctuation">,</span> partition<span class="token punctuation">,</span> topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token punctuation">}</span> <span class="token keyword">else</span> <span class="token punctuation">{</span>
                    <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">trace</span><span class="token punctuation">(</span><span class="token string">&quot;Requesting metadata update for topic {}.&quot;</span><span class="token punctuation">,</span> topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token punctuation">}</span>

                <span class="token keyword">this</span><span class="token punctuation">.</span>metadata<span class="token punctuation">.</span><span class="token function">add</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token keyword">int</span> version <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>metadata<span class="token punctuation">.</span><span class="token function">requestUpdate</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token keyword">this</span><span class="token punctuation">.</span>sender<span class="token punctuation">.</span><span class="token function">wakeup</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>

                <span class="token keyword">try</span> <span class="token punctuation">{</span>
                    <span class="token keyword">this</span><span class="token punctuation">.</span>metadata<span class="token punctuation">.</span><span class="token function">awaitUpdate</span><span class="token punctuation">(</span>version<span class="token punctuation">,</span> remainingWaitMs<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token punctuation">}</span> <span class="token keyword">catch</span> <span class="token punctuation">(</span><span class="token class-name">TimeoutException</span> var15<span class="token punctuation">)</span> <span class="token punctuation">{</span>
                    <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">TimeoutException</span><span class="token punctuation">(</span><span class="token class-name">String</span><span class="token punctuation">.</span><span class="token function">format</span><span class="token punctuation">(</span><span class="token string">&quot;Topic %s not present in metadata after %d ms.&quot;</span><span class="token punctuation">,</span> topic<span class="token punctuation">,</span> maxWaitMs<span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token punctuation">}</span>

                cluster <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>metadata<span class="token punctuation">.</span><span class="token function">fetch</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
                elapsed <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>time<span class="token punctuation">.</span><span class="token function">milliseconds</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">-</span> begin<span class="token punctuation">;</span>
                <span class="token keyword">if</span> <span class="token punctuation">(</span>elapsed <span class="token operator">&gt;=</span> maxWaitMs<span class="token punctuation">)</span> <span class="token punctuation">{</span>
                    <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">TimeoutException</span><span class="token punctuation">(</span>partitionsCount <span class="token operator">==</span> <span class="token keyword">null</span> <span class="token operator">?</span> <span class="token class-name">String</span><span class="token punctuation">.</span><span class="token function">format</span><span class="token punctuation">(</span><span class="token string">&quot;Topic %s not present in metadata after %d ms.&quot;</span><span class="token punctuation">,</span> topic<span class="token punctuation">,</span> maxWaitMs<span class="token punctuation">)</span> <span class="token operator">:</span> <span class="token class-name">String</span><span class="token punctuation">.</span><span class="token function">format</span><span class="token punctuation">(</span><span class="token string">&quot;Partition %d of topic %s with partition count %d is not present in metadata after %d ms.&quot;</span><span class="token punctuation">,</span> partition<span class="token punctuation">,</span> topic<span class="token punctuation">,</span> partitionsCount<span class="token punctuation">,</span> maxWaitMs<span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token punctuation">}</span>

                <span class="token keyword">if</span> <span class="token punctuation">(</span>cluster<span class="token punctuation">.</span><span class="token function">unauthorizedTopics</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">contains</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                    <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">TopicAuthorizationException</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token punctuation">}</span>

                <span class="token keyword">if</span> <span class="token punctuation">(</span>cluster<span class="token punctuation">.</span><span class="token function">invalidTopics</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">contains</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                    <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">InvalidTopicException</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token punctuation">}</span>

                remainingWaitMs <span class="token operator">=</span> maxWaitMs <span class="token operator">-</span> elapsed<span class="token punctuation">;</span>
                partitionsCount <span class="token operator">=</span> cluster<span class="token punctuation">.</span><span class="token function">partitionCountForTopic</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span> <span class="token keyword">while</span><span class="token punctuation">(</span>partitionsCount <span class="token operator">==</span> <span class="token keyword">null</span> <span class="token operator">||</span> partition <span class="token operator">!=</span> <span class="token keyword">null</span> <span class="token operator">&amp;&amp;</span> partition <span class="token operator">&gt;=</span> partitionsCount<span class="token punctuation">)</span><span class="token punctuation">;</span>

            <span class="token keyword">return</span> <span class="token keyword">new</span> <span class="token class-name">KafkaProducer</span><span class="token punctuation">.</span><span class="token class-name">ClusterAndWaitTime</span><span class="token punctuation">(</span>cluster<span class="token punctuation">,</span> elapsed<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span> <span class="token keyword">else</span> <span class="token punctuation">{</span>
            <span class="token keyword">return</span> <span class="token keyword">new</span> <span class="token class-name">KafkaProducer</span><span class="token punctuation">.</span><span class="token class-name">ClusterAndWaitTime</span><span class="token punctuation">(</span>cluster<span class="token punctuation">,</span> <span class="token number">0L</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span>
<span class="token punctuation">}</span>
</code></pre></div><h4 id="确定partition值"><a href="#确定partition值" class="header-anchor">#</a> 确定partition值</h4> <p>关于 partition 值的计算，分为三种情况：</p> <ol><li>指明 partition 的情况下，直接将指定的值直接作为 partiton 值；</li> <li>没有指明 partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值；</li> <li>既没有 partition 值又没有 key 值的情况下，第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition 值，也就是常说的轮询算法。</li></ol> <div class="language-java extra-class"><pre class="language-java"><code><span class="token keyword">private</span> <span class="token keyword">int</span> <span class="token function">partition</span><span class="token punctuation">(</span><span class="token class-name">ProducerRecord</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">K</span><span class="token punctuation">,</span> <span class="token class-name">V</span><span class="token punctuation">&gt;</span></span> <span class="token keyword">record</span><span class="token punctuation">,</span> <span class="token keyword">byte</span><span class="token punctuation">[</span><span class="token punctuation">]</span> serializedKey<span class="token punctuation">,</span> <span class="token keyword">byte</span><span class="token punctuation">[</span><span class="token punctuation">]</span> serializedValue<span class="token punctuation">,</span> <span class="token class-name">Cluster</span> cluster<span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token class-name">Integer</span> partition <span class="token operator">=</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">partition</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">return</span> partition <span class="token operator">!=</span> <span class="token keyword">null</span> <span class="token operator">?</span> partition <span class="token operator">:</span> <span class="token keyword">this</span><span class="token punctuation">.</span>partitioner<span class="token punctuation">.</span><span class="token function">partition</span><span class="token punctuation">(</span><span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">topic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">key</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> serializedKey<span class="token punctuation">,</span> <span class="token keyword">record</span><span class="token punctuation">.</span><span class="token function">value</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> serializedValue<span class="token punctuation">,</span> cluster<span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre></div><p><strong>DefaultPartitioner 的实现</strong></p> <div class="language-java extra-class"><pre class="language-java"><code><span class="token keyword">public</span> <span class="token keyword">int</span> <span class="token function">partition</span><span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Object</span> key<span class="token punctuation">,</span> <span class="token keyword">byte</span><span class="token punctuation">[</span><span class="token punctuation">]</span> keyBytes<span class="token punctuation">,</span> <span class="token class-name">Object</span> value<span class="token punctuation">,</span> <span class="token keyword">byte</span><span class="token punctuation">[</span><span class="token punctuation">]</span> valueBytes<span class="token punctuation">,</span> <span class="token class-name">Cluster</span> cluster<span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token class-name">List</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">PartitionInfo</span><span class="token punctuation">&gt;</span></span> partitions <span class="token operator">=</span> cluster<span class="token punctuation">.</span><span class="token function">partitionsForTopic</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">int</span> numPartitions <span class="token operator">=</span> partitions<span class="token punctuation">.</span><span class="token function">size</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">if</span> <span class="token punctuation">(</span>keyBytes <span class="token operator">==</span> <span class="token keyword">null</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token comment">//不指定key时,根据 topic 获取对应的整数变量</span>
        <span class="token keyword">int</span> nextValue <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">nextValue</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token class-name">List</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">PartitionInfo</span><span class="token punctuation">&gt;</span></span> availablePartitions <span class="token operator">=</span> cluster<span class="token punctuation">.</span><span class="token function">availablePartitionsForTopic</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">if</span> <span class="token punctuation">(</span>availablePartitions<span class="token punctuation">.</span><span class="token function">size</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">&gt;</span> <span class="token number">0</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">int</span> part <span class="token operator">=</span> <span class="token class-name">Utils</span><span class="token punctuation">.</span><span class="token function">toPositive</span><span class="token punctuation">(</span>nextValue<span class="token punctuation">)</span> <span class="token operator">%</span> availablePartitions<span class="token punctuation">.</span><span class="token function">size</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token keyword">return</span> <span class="token punctuation">(</span><span class="token punctuation">(</span><span class="token class-name">PartitionInfo</span><span class="token punctuation">)</span>availablePartitions<span class="token punctuation">.</span><span class="token function">get</span><span class="token punctuation">(</span>part<span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">partition</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span> <span class="token keyword">else</span> <span class="token punctuation">{</span>
            <span class="token keyword">return</span> <span class="token class-name">Utils</span><span class="token punctuation">.</span><span class="token function">toPositive</span><span class="token punctuation">(</span>nextValue<span class="token punctuation">)</span> <span class="token operator">%</span> numPartitions<span class="token punctuation">;</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span> <span class="token keyword">else</span> <span class="token punctuation">{</span>
        <span class="token comment">//指定key时，通过key进行hash运算然后对该topic分区总数取余</span>
        <span class="token keyword">return</span> <span class="token class-name">Utils</span><span class="token punctuation">.</span><span class="token function">toPositive</span><span class="token punctuation">(</span><span class="token class-name">Utils</span><span class="token punctuation">.</span><span class="token function">murmur2</span><span class="token punctuation">(</span>keyBytes<span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token operator">%</span> numPartitions<span class="token punctuation">;</span>
    <span class="token punctuation">}</span>
<span class="token punctuation">}</span>

<span class="token keyword">private</span> <span class="token keyword">int</span> <span class="token function">nextValue</span><span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token class-name">AtomicInteger</span> counter <span class="token operator">=</span> <span class="token punctuation">(</span><span class="token class-name">AtomicInteger</span><span class="token punctuation">)</span><span class="token keyword">this</span><span class="token punctuation">.</span>topicCounterMap<span class="token punctuation">.</span><span class="token function">get</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">null</span> <span class="token operator">==</span> counter<span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token comment">//第一次随机生成一个数</span>
        counter <span class="token operator">=</span> <span class="token keyword">new</span> <span class="token class-name">AtomicInteger</span><span class="token punctuation">(</span><span class="token class-name">ThreadLocalRandom</span><span class="token punctuation">.</span><span class="token function">current</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">nextInt</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token class-name">AtomicInteger</span> currentCounter <span class="token operator">=</span> <span class="token punctuation">(</span><span class="token class-name">AtomicInteger</span><span class="token punctuation">)</span><span class="token keyword">this</span><span class="token punctuation">.</span>topicCounterMap<span class="token punctuation">.</span><span class="token function">putIfAbsent</span><span class="token punctuation">(</span>topic<span class="token punctuation">,</span> counter<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">if</span> <span class="token punctuation">(</span>currentCounter <span class="token operator">!=</span> <span class="token keyword">null</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
            counter <span class="token operator">=</span> currentCounter<span class="token punctuation">;</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span>
	<span class="token comment">//以后每次递增</span>
    <span class="token keyword">return</span> counter<span class="token punctuation">.</span><span class="token function">getAndIncrement</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre></div><h4 id="估算消息大小并检查"><a href="#估算消息大小并检查" class="header-anchor">#</a> 估算消息大小并检查</h4> <p>校验消息是为了防止消息过大或者数量过多，导致内存异常，具体实现</p> <div class="language-java extra-class"><pre class="language-java"><code><span class="token comment">//估算消息大小</span>
<span class="token keyword">int</span> serializedSize <span class="token operator">=</span> <span class="token class-name">AbstractRecords</span><span class="token punctuation">.</span><span class="token function">estimateSizeInBytesUpperBound</span><span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>apiVersions<span class="token punctuation">.</span><span class="token function">maxUsableProduceMagic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>compressionType<span class="token punctuation">,</span> serializedKey<span class="token punctuation">,</span> serializedValue<span class="token punctuation">,</span> headers<span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token comment">//确保消息大小在有效范围内</span>
<span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">ensureValidRecordSize</span><span class="token punctuation">(</span>serializedSize<span class="token punctuation">)</span><span class="token punctuation">;</span>

    <span class="token keyword">private</span> <span class="token keyword">void</span> <span class="token function">ensureValidRecordSize</span><span class="token punctuation">(</span><span class="token keyword">int</span> size<span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token comment">//是否超过最大请求的限制,如果超过抛出异常</span>
        <span class="token keyword">if</span> <span class="token punctuation">(</span>size <span class="token operator">&gt;</span> <span class="token keyword">this</span><span class="token punctuation">.</span>maxRequestSize<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">RecordTooLargeException</span><span class="token punctuation">(</span><span class="token string">&quot;The message is &quot;</span> <span class="token operator">+</span> size <span class="token operator">+</span> <span class="token string">&quot; bytes when serialized which is larger than the maximum request size you have configured with the &quot;</span> <span class="token operator">+</span> <span class="token string">&quot;max.request.size&quot;</span> <span class="token operator">+</span> <span class="token string">&quot; configuration.&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token comment">//是否超过最大内存缓冲池大小,如果超过抛出异常</span>
        <span class="token punctuation">}</span> <span class="token keyword">else</span> <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token punctuation">(</span><span class="token keyword">long</span><span class="token punctuation">)</span>size <span class="token operator">&gt;</span> <span class="token keyword">this</span><span class="token punctuation">.</span>totalMemorySize<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">RecordTooLargeException</span><span class="token punctuation">(</span><span class="token string">&quot;The message is &quot;</span> <span class="token operator">+</span> size <span class="token operator">+</span> <span class="token string">&quot; bytes when serialized which is larger than the total memory buffer you have configured with the &quot;</span> <span class="token operator">+</span> <span class="token string">&quot;buffer.memory&quot;</span> <span class="token operator">+</span> <span class="token string">&quot; configuration.&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span>
</code></pre></div><h4 id="向-accumulator-写数据"><a href="#向-accumulator-写数据" class="header-anchor">#</a> 向 accumulator 写数据</h4> <div class="language-java extra-class"><pre class="language-java"><code><span class="token keyword">public</span> <span class="token class-name">RecordAccumulator</span><span class="token punctuation">.</span><span class="token class-name">RecordAppendResult</span> <span class="token function">append</span><span class="token punctuation">(</span><span class="token class-name">TopicPartition</span> tp<span class="token punctuation">,</span> <span class="token keyword">long</span> timestamp<span class="token punctuation">,</span> <span class="token keyword">byte</span><span class="token punctuation">[</span><span class="token punctuation">]</span> key<span class="token punctuation">,</span> <span class="token keyword">byte</span><span class="token punctuation">[</span><span class="token punctuation">]</span> value<span class="token punctuation">,</span> <span class="token class-name">Header</span><span class="token punctuation">[</span><span class="token punctuation">]</span> headers<span class="token punctuation">,</span> <span class="token class-name">Callback</span> callback<span class="token punctuation">,</span> <span class="token keyword">long</span> maxTimeToBlock<span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">InterruptedException</span> <span class="token punctuation">{</span>
    <span class="token keyword">this</span><span class="token punctuation">.</span>appendsInProgress<span class="token punctuation">.</span><span class="token function">incrementAndGet</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token class-name">ByteBuffer</span> buffer <span class="token operator">=</span> <span class="token keyword">null</span><span class="token punctuation">;</span>
    <span class="token keyword">if</span> <span class="token punctuation">(</span>headers <span class="token operator">==</span> <span class="token keyword">null</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        headers <span class="token operator">=</span> <span class="token class-name">Record</span><span class="token punctuation">.</span>EMPTY_HEADERS<span class="token punctuation">;</span>
    <span class="token punctuation">}</span>

    <span class="token class-name">RecordAccumulator</span><span class="token punctuation">.</span><span class="token class-name">RecordAppendResult</span> var16<span class="token punctuation">;</span>
    <span class="token keyword">try</span> <span class="token punctuation">{</span>
        <span class="token comment">// 当前tp有对应queue时直接返回，否则新建一个返回</span>
        <span class="token class-name">Deque</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">ProducerBatch</span><span class="token punctuation">&gt;</span></span> dq <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">getOrCreateDeque</span><span class="token punctuation">(</span>tp<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token comment">// 在对一个 queue 进行操作时,会保证线程安全</span>
        <span class="token keyword">synchronized</span><span class="token punctuation">(</span>dq<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>closed<span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">KafkaException</span><span class="token punctuation">(</span><span class="token string">&quot;Producer closed while send in progress&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span>

            <span class="token class-name">RecordAccumulator</span><span class="token punctuation">.</span><span class="token class-name">RecordAppendResult</span> appendResult <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">tryAppend</span><span class="token punctuation">(</span>timestamp<span class="token punctuation">,</span> key<span class="token punctuation">,</span> value<span class="token punctuation">,</span> headers<span class="token punctuation">,</span> callback<span class="token punctuation">,</span> dq<span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token keyword">if</span> <span class="token punctuation">(</span>appendResult <span class="token operator">!=</span> <span class="token keyword">null</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token class-name">RecordAccumulator</span><span class="token punctuation">.</span><span class="token class-name">RecordAppendResult</span> var14 <span class="token operator">=</span> appendResult<span class="token punctuation">;</span>
                <span class="token keyword">return</span> var14<span class="token punctuation">;</span>
            <span class="token punctuation">}</span>
        <span class="token punctuation">}</span>
		<span class="token comment">// 为 topic-partition 创建一个新的 RecordBatch</span>
        <span class="token keyword">byte</span> maxUsableMagic <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>apiVersions<span class="token punctuation">.</span><span class="token function">maxUsableProduceMagic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">int</span> size <span class="token operator">=</span> <span class="token class-name">Math</span><span class="token punctuation">.</span><span class="token function">max</span><span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>batchSize<span class="token punctuation">,</span> <span class="token class-name">AbstractRecords</span><span class="token punctuation">.</span><span class="token function">estimateSizeInBytesUpperBound</span><span class="token punctuation">(</span>maxUsableMagic<span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>compression<span class="token punctuation">,</span> key<span class="token punctuation">,</span> value<span class="token punctuation">,</span> headers<span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">trace</span><span class="token punctuation">(</span><span class="token string">&quot;Allocating a new {} byte message buffer for topic {} partition {}&quot;</span><span class="token punctuation">,</span> <span class="token keyword">new</span> <span class="token class-name">Object</span><span class="token punctuation">[</span><span class="token punctuation">]</span><span class="token punctuation">{</span>size<span class="token punctuation">,</span> tp<span class="token punctuation">.</span><span class="token function">topic</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> tp<span class="token punctuation">.</span><span class="token function">partition</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">}</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token comment">// 给这个 RecordBatch 初始化一个 buffer</span>
        buffer <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>free<span class="token punctuation">.</span><span class="token function">allocate</span><span class="token punctuation">(</span>size<span class="token punctuation">,</span> maxTimeToBlock<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">synchronized</span><span class="token punctuation">(</span>dq<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>closed<span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token keyword">throw</span> <span class="token keyword">new</span> <span class="token class-name">KafkaException</span><span class="token punctuation">(</span><span class="token string">&quot;Producer closed while send in progress&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span>

            <span class="token class-name">RecordAccumulator</span><span class="token punctuation">.</span><span class="token class-name">RecordAppendResult</span> appendResult <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">tryAppend</span><span class="token punctuation">(</span>timestamp<span class="token punctuation">,</span> key<span class="token punctuation">,</span> value<span class="token punctuation">,</span> headers<span class="token punctuation">,</span> callback<span class="token punctuation">,</span> dq<span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token keyword">if</span> <span class="token punctuation">(</span>appendResult <span class="token operator">==</span> <span class="token keyword">null</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token comment">// 给 topic-partition 创建一个 RecordBatch</span>
                <span class="token class-name">MemoryRecordsBuilder</span> recordsBuilder <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">recordsBuilder</span><span class="token punctuation">(</span>buffer<span class="token punctuation">,</span> maxUsableMagic<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token class-name">ProducerBatch</span> batch <span class="token operator">=</span> <span class="token keyword">new</span> <span class="token class-name">ProducerBatch</span><span class="token punctuation">(</span>tp<span class="token punctuation">,</span> recordsBuilder<span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>time<span class="token punctuation">.</span><span class="token function">milliseconds</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token comment">// 向新的 RecordBatch 中追加数据</span>
                <span class="token class-name">FutureRecordMetadata</span> future <span class="token operator">=</span> <span class="token punctuation">(</span><span class="token class-name">FutureRecordMetadata</span><span class="token punctuation">)</span><span class="token class-name">Utils</span><span class="token punctuation">.</span><span class="token function">notNull</span><span class="token punctuation">(</span>batch<span class="token punctuation">.</span><span class="token function">tryAppend</span><span class="token punctuation">(</span>timestamp<span class="token punctuation">,</span> key<span class="token punctuation">,</span> value<span class="token punctuation">,</span> headers<span class="token punctuation">,</span> callback<span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>time<span class="token punctuation">.</span><span class="token function">milliseconds</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token comment">// 将 RecordBatch 添加到对应的 queue 中</span>
                dq<span class="token punctuation">.</span><span class="token function">addLast</span><span class="token punctuation">(</span>batch<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token comment">// 向未 ack 的 batch 集合添加这个 batch</span>
                <span class="token keyword">this</span><span class="token punctuation">.</span>incomplete<span class="token punctuation">.</span><span class="token function">add</span><span class="token punctuation">(</span>batch<span class="token punctuation">)</span><span class="token punctuation">;</span>
                buffer <span class="token operator">=</span> <span class="token keyword">null</span><span class="token punctuation">;</span>
                <span class="token comment">// 如果 dp.size()&gt;1 就证明这个 queue 有一个 batch 是可以发送了</span>
                <span class="token class-name">RecordAccumulator</span><span class="token punctuation">.</span><span class="token class-name">RecordAppendResult</span> var19 <span class="token operator">=</span> <span class="token keyword">new</span> <span class="token class-name">RecordAccumulator</span><span class="token punctuation">.</span><span class="token class-name">RecordAppendResult</span><span class="token punctuation">(</span>future<span class="token punctuation">,</span> dq<span class="token punctuation">.</span><span class="token function">size</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">&gt;</span> <span class="token number">1</span> <span class="token operator">||</span> batch<span class="token punctuation">.</span><span class="token function">isFull</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token boolean">true</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token keyword">return</span> var19<span class="token punctuation">;</span>
            <span class="token punctuation">}</span>

            var16 <span class="token operator">=</span> appendResult<span class="token punctuation">;</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span> <span class="token keyword">finally</span> <span class="token punctuation">{</span>
        <span class="token keyword">if</span> <span class="token punctuation">(</span>buffer <span class="token operator">!=</span> <span class="token keyword">null</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">this</span><span class="token punctuation">.</span>free<span class="token punctuation">.</span><span class="token function">deallocate</span><span class="token punctuation">(</span>buffer<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>

        <span class="token keyword">this</span><span class="token punctuation">.</span>appendsInProgress<span class="token punctuation">.</span><span class="token function">decrementAndGet</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span>

    <span class="token keyword">return</span> var16<span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre></div><h4 id="使用sender线程批量发送"><a href="#使用sender线程批量发送" class="header-anchor">#</a> 使用sender线程批量发送</h4> <div class="language-java extra-class"><pre class="language-java"><code><span class="token keyword">public</span> <span class="token keyword">void</span> <span class="token function">run</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">debug</span><span class="token punctuation">(</span><span class="token string">&quot;Starting Kafka producer I/O thread.&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>

    <span class="token keyword">while</span><span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>running<span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token keyword">try</span> <span class="token punctuation">{</span>
            <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">run</span><span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>time<span class="token punctuation">.</span><span class="token function">milliseconds</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span> <span class="token keyword">catch</span> <span class="token punctuation">(</span><span class="token class-name">Exception</span> var4<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">error</span><span class="token punctuation">(</span><span class="token string">&quot;Uncaught error in kafka producer I/O thread: &quot;</span><span class="token punctuation">,</span> var4<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span>

    <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">debug</span><span class="token punctuation">(</span><span class="token string">&quot;Beginning shutdown of Kafka producer I/O thread, sending remaining records.&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
	<span class="token comment">// 累加器中可能仍然有请求，或者等待确认,等待他们完成就停止接受请求</span>
    <span class="token keyword">while</span><span class="token punctuation">(</span><span class="token operator">!</span><span class="token keyword">this</span><span class="token punctuation">.</span>forceClose <span class="token operator">&amp;&amp;</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>accumulator<span class="token punctuation">.</span><span class="token function">hasUndrained</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">||</span> <span class="token keyword">this</span><span class="token punctuation">.</span>client<span class="token punctuation">.</span><span class="token function">inFlightRequestCount</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">&gt;</span> <span class="token number">0</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token comment">//......</span>
    <span class="token punctuation">}</span>

    <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>forceClose<span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token comment">//......</span>
    <span class="token punctuation">}</span>

    <span class="token keyword">try</span> <span class="token punctuation">{</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span>client<span class="token punctuation">.</span><span class="token function">close</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span> <span class="token keyword">catch</span> <span class="token punctuation">(</span><span class="token class-name">Exception</span> var2<span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">error</span><span class="token punctuation">(</span><span class="token string">&quot;Failed to close network client&quot;</span><span class="token punctuation">,</span> var2<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span>

    <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">debug</span><span class="token punctuation">(</span><span class="token string">&quot;Shutdown of Kafka producer I/O thread has completed.&quot;</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>

<span class="token keyword">void</span> <span class="token function">run</span><span class="token punctuation">(</span><span class="token keyword">long</span> now<span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token comment">//首先还是判断是否使用事务发送</span>
    <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager <span class="token operator">!=</span> <span class="token keyword">null</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token keyword">try</span> <span class="token punctuation">{</span>
            <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">shouldResetProducerStateAfterResolvingSequences</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">resetProducerId</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span>

            <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token operator">!</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">isTransactional</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">maybeWaitForProducerId</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span> <span class="token keyword">else</span> <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">hasUnresolvedSequences</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">&amp;&amp;</span> <span class="token operator">!</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">hasFatalError</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">transitionToFatalError</span><span class="token punctuation">(</span><span class="token keyword">new</span> <span class="token class-name">KafkaException</span><span class="token punctuation">(</span><span class="token string">&quot;The client hasn't received acknowledgment for some previously sent messages and can no longer retry them. It isn't safe to continue.&quot;</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span> <span class="token keyword">else</span> <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">hasInFlightTransactionalRequest</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">||</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">maybeSendTransactionalRequest</span><span class="token punctuation">(</span>now<span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token keyword">this</span><span class="token punctuation">.</span>client<span class="token punctuation">.</span><span class="token function">poll</span><span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>retryBackoffMs<span class="token punctuation">,</span> now<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token keyword">return</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span>
			<span class="token comment">//如果事务管理器转态错误 或者producer没有id 将停止进行发送</span>
            <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">hasFatalError</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">||</span> <span class="token operator">!</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">hasProducerId</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token class-name">RuntimeException</span> lastError <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">lastError</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token keyword">if</span> <span class="token punctuation">(</span>lastError <span class="token operator">!=</span> <span class="token keyword">null</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                    <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">maybeAbortBatches</span><span class="token punctuation">(</span>lastError<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token punctuation">}</span>

                <span class="token keyword">this</span><span class="token punctuation">.</span>client<span class="token punctuation">.</span><span class="token function">poll</span><span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>retryBackoffMs<span class="token punctuation">,</span> now<span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token keyword">return</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span>

            <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">hasAbortableError</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                <span class="token keyword">this</span><span class="token punctuation">.</span>accumulator<span class="token punctuation">.</span><span class="token function">abortUndrainedBatches</span><span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">lastError</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span>
        <span class="token punctuation">}</span> <span class="token keyword">catch</span> <span class="token punctuation">(</span><span class="token class-name">AuthenticationException</span> var5<span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">trace</span><span class="token punctuation">(</span><span class="token string">&quot;Authentication exception while processing transactional request: {}&quot;</span><span class="token punctuation">,</span> var5<span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">authenticationFailed</span><span class="token punctuation">(</span>var5<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span>
	<span class="token comment">//发送Producer数据</span>
    <span class="token keyword">long</span> pollTimeout <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">sendProducerData</span><span class="token punctuation">(</span>now<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">this</span><span class="token punctuation">.</span>client<span class="token punctuation">.</span><span class="token function">poll</span><span class="token punctuation">(</span>pollTimeout<span class="token punctuation">,</span> now<span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>

<span class="token keyword">private</span> <span class="token keyword">long</span> <span class="token function">sendProducerData</span><span class="token punctuation">(</span><span class="token keyword">long</span> now<span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token class-name">Cluster</span> cluster <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>metadata<span class="token punctuation">.</span><span class="token function">fetch</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token comment">// 获取准备发送数据的分区列表</span>
    <span class="token class-name">ReadyCheckResult</span> result <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>accumulator<span class="token punctuation">.</span><span class="token function">ready</span><span class="token punctuation">(</span>cluster<span class="token punctuation">,</span> now<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token class-name">Iterator</span> iter<span class="token punctuation">;</span>
    <span class="token comment">//如果有分区leader信息是未知的,那么就强制更新metadata</span>
    <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token operator">!</span>result<span class="token punctuation">.</span>unknownLeaderTopics<span class="token punctuation">.</span><span class="token function">isEmpty</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        iter <span class="token operator">=</span> result<span class="token punctuation">.</span>unknownLeaderTopics<span class="token punctuation">.</span><span class="token function">iterator</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>

        <span class="token keyword">while</span><span class="token punctuation">(</span>iter<span class="token punctuation">.</span><span class="token function">hasNext</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token class-name">String</span> topic <span class="token operator">=</span> <span class="token punctuation">(</span><span class="token class-name">String</span><span class="token punctuation">)</span>iter<span class="token punctuation">.</span><span class="token function">next</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token keyword">this</span><span class="token punctuation">.</span>metadata<span class="token punctuation">.</span><span class="token function">add</span><span class="token punctuation">(</span>topic<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>

        <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">debug</span><span class="token punctuation">(</span><span class="token string">&quot;Requesting metadata update due to unknown leader topics from the batched records: {}&quot;</span><span class="token punctuation">,</span> result<span class="token punctuation">.</span>unknownLeaderTopics<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span>metadata<span class="token punctuation">.</span><span class="token function">requestUpdate</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span>

    iter <span class="token operator">=</span> result<span class="token punctuation">.</span>readyNodes<span class="token punctuation">.</span><span class="token function">iterator</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">long</span> notReadyTimeout <span class="token operator">=</span> <span class="token number">9223372036854775807L</span><span class="token punctuation">;</span>

    <span class="token comment">// 移除没有准备好发送的Node</span>
    <span class="token keyword">while</span><span class="token punctuation">(</span>iter<span class="token punctuation">.</span><span class="token function">hasNext</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token class-name">Node</span> node <span class="token operator">=</span> <span class="token punctuation">(</span><span class="token class-name">Node</span><span class="token punctuation">)</span>iter<span class="token punctuation">.</span><span class="token function">next</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token operator">!</span><span class="token keyword">this</span><span class="token punctuation">.</span>client<span class="token punctuation">.</span><span class="token function">ready</span><span class="token punctuation">(</span>node<span class="token punctuation">,</span> now<span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
            iter<span class="token punctuation">.</span><span class="token function">remove</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            notReadyTimeout <span class="token operator">=</span> <span class="token class-name">Math</span><span class="token punctuation">.</span><span class="token function">min</span><span class="token punctuation">(</span>notReadyTimeout<span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>client<span class="token punctuation">.</span><span class="token function">pollDelayMs</span><span class="token punctuation">(</span>node<span class="token punctuation">,</span> now<span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span>
	<span class="token comment">//创建Producer请求内容</span>
    <span class="token class-name">Map</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">Integer</span><span class="token punctuation">,</span> <span class="token class-name">List</span><span class="token punctuation">&lt;</span><span class="token class-name">ProducerBatch</span><span class="token punctuation">&gt;</span><span class="token punctuation">&gt;</span></span> batches <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>accumulator<span class="token punctuation">.</span><span class="token function">drain</span><span class="token punctuation">(</span>cluster<span class="token punctuation">,</span> result<span class="token punctuation">.</span>readyNodes<span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>maxRequestSize<span class="token punctuation">,</span> now<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">addToInflightBatches</span><span class="token punctuation">(</span>batches<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token class-name">List</span> expiredBatches<span class="token punctuation">;</span>
    <span class="token class-name">Iterator</span> var11<span class="token punctuation">;</span>
    <span class="token class-name">ProducerBatch</span> expiredBatch<span class="token punctuation">;</span>
    <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>guaranteeMessageOrder<span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token class-name">Iterator</span> var9 <span class="token operator">=</span> batches<span class="token punctuation">.</span><span class="token function">values</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">.</span><span class="token function">iterator</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>

        <span class="token keyword">while</span><span class="token punctuation">(</span>var9<span class="token punctuation">.</span><span class="token function">hasNext</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
            expiredBatches <span class="token operator">=</span> <span class="token punctuation">(</span><span class="token class-name">List</span><span class="token punctuation">)</span>var9<span class="token punctuation">.</span><span class="token function">next</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
            var11 <span class="token operator">=</span> expiredBatches<span class="token punctuation">.</span><span class="token function">iterator</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>

            <span class="token keyword">while</span><span class="token punctuation">(</span>var11<span class="token punctuation">.</span><span class="token function">hasNext</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
                expiredBatch <span class="token operator">=</span> <span class="token punctuation">(</span><span class="token class-name">ProducerBatch</span><span class="token punctuation">)</span>var11<span class="token punctuation">.</span><span class="token function">next</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
                <span class="token keyword">this</span><span class="token punctuation">.</span>accumulator<span class="token punctuation">.</span><span class="token function">mutePartition</span><span class="token punctuation">(</span>expiredBatch<span class="token punctuation">.</span>topicPartition<span class="token punctuation">)</span><span class="token punctuation">;</span>
            <span class="token punctuation">}</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span>

    <span class="token keyword">this</span><span class="token punctuation">.</span>accumulator<span class="token punctuation">.</span><span class="token function">resetNextBatchExpiryTime</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token class-name">List</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">ProducerBatch</span><span class="token punctuation">&gt;</span></span> expiredInflightBatches <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">getExpiredInflightBatches</span><span class="token punctuation">(</span>now<span class="token punctuation">)</span><span class="token punctuation">;</span>
    expiredBatches <span class="token operator">=</span> <span class="token keyword">this</span><span class="token punctuation">.</span>accumulator<span class="token punctuation">.</span><span class="token function">expiredBatches</span><span class="token punctuation">(</span>now<span class="token punctuation">)</span><span class="token punctuation">;</span>
    expiredBatches<span class="token punctuation">.</span><span class="token function">addAll</span><span class="token punctuation">(</span>expiredInflightBatches<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token operator">!</span>expiredBatches<span class="token punctuation">.</span><span class="token function">isEmpty</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">trace</span><span class="token punctuation">(</span><span class="token string">&quot;Expired {} batches in accumulator&quot;</span><span class="token punctuation">,</span> expiredBatches<span class="token punctuation">.</span><span class="token function">size</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span>

    var11 <span class="token operator">=</span> expiredBatches<span class="token punctuation">.</span><span class="token function">iterator</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>

    <span class="token keyword">while</span><span class="token punctuation">(</span>var11<span class="token punctuation">.</span><span class="token function">hasNext</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        expiredBatch <span class="token operator">=</span> <span class="token punctuation">(</span><span class="token class-name">ProducerBatch</span><span class="token punctuation">)</span>var11<span class="token punctuation">.</span><span class="token function">next</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token class-name">String</span> errorMessage <span class="token operator">=</span> <span class="token string">&quot;Expiring &quot;</span> <span class="token operator">+</span> expiredBatch<span class="token punctuation">.</span>recordCount <span class="token operator">+</span> <span class="token string">&quot; record(s) for &quot;</span> <span class="token operator">+</span> expiredBatch<span class="token punctuation">.</span>topicPartition <span class="token operator">+</span> <span class="token string">&quot;:&quot;</span> <span class="token operator">+</span> <span class="token punctuation">(</span>now <span class="token operator">-</span> expiredBatch<span class="token punctuation">.</span>createdMs<span class="token punctuation">)</span> <span class="token operator">+</span> <span class="token string">&quot; ms has passed since batch creation&quot;</span><span class="token punctuation">;</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">failBatch</span><span class="token punctuation">(</span>expiredBatch<span class="token punctuation">,</span> <span class="token operator">-</span><span class="token number">1L</span><span class="token punctuation">,</span> <span class="token operator">-</span><span class="token number">1L</span><span class="token punctuation">,</span> <span class="token keyword">new</span> <span class="token class-name">TimeoutException</span><span class="token punctuation">(</span>errorMessage<span class="token punctuation">)</span><span class="token punctuation">,</span> <span class="token boolean">false</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager <span class="token operator">!=</span> <span class="token keyword">null</span> <span class="token operator">&amp;&amp;</span> expiredBatch<span class="token punctuation">.</span><span class="token function">inRetry</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">this</span><span class="token punctuation">.</span>transactionManager<span class="token punctuation">.</span><span class="token function">markSequenceUnresolved</span><span class="token punctuation">(</span>expiredBatch<span class="token punctuation">.</span>topicPartition<span class="token punctuation">)</span><span class="token punctuation">;</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span>

    <span class="token keyword">this</span><span class="token punctuation">.</span>sensors<span class="token punctuation">.</span><span class="token function">updateProduceRequestMetrics</span><span class="token punctuation">(</span>batches<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">long</span> pollTimeout <span class="token operator">=</span> <span class="token class-name">Math</span><span class="token punctuation">.</span><span class="token function">min</span><span class="token punctuation">(</span>result<span class="token punctuation">.</span>nextReadyCheckDelayMs<span class="token punctuation">,</span> notReadyTimeout<span class="token punctuation">)</span><span class="token punctuation">;</span>
    pollTimeout <span class="token operator">=</span> <span class="token class-name">Math</span><span class="token punctuation">.</span><span class="token function">min</span><span class="token punctuation">(</span>pollTimeout<span class="token punctuation">,</span> <span class="token keyword">this</span><span class="token punctuation">.</span>accumulator<span class="token punctuation">.</span><span class="token function">nextExpiryTimeMs</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token operator">-</span> now<span class="token punctuation">)</span><span class="token punctuation">;</span>
    pollTimeout <span class="token operator">=</span> <span class="token class-name">Math</span><span class="token punctuation">.</span><span class="token function">max</span><span class="token punctuation">(</span>pollTimeout<span class="token punctuation">,</span> <span class="token number">0L</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">if</span> <span class="token punctuation">(</span><span class="token operator">!</span>result<span class="token punctuation">.</span>readyNodes<span class="token punctuation">.</span><span class="token function">isEmpty</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
        <span class="token keyword">this</span><span class="token punctuation">.</span>log<span class="token punctuation">.</span><span class="token function">trace</span><span class="token punctuation">(</span><span class="token string">&quot;Nodes with data ready to send: {}&quot;</span><span class="token punctuation">,</span> result<span class="token punctuation">.</span>readyNodes<span class="token punctuation">)</span><span class="token punctuation">;</span>
        pollTimeout <span class="token operator">=</span> <span class="token number">0L</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span>

    <span class="token keyword">this</span><span class="token punctuation">.</span><span class="token function">sendProduceRequests</span><span class="token punctuation">(</span>batches<span class="token punctuation">,</span> now<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token keyword">return</span> pollTimeout<span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre></div><p><img src="https://static01.imgkr.com/temp/7e16aac5204940cb9033ef7c93512a16.png" alt=""></p> <p><img src="https://static01.imgkr.com/temp/e414ef7c122f4c1cae598047d96d3f49.png" alt=""></p> <p>之后在IDEA中 , 双击shift , 调出全局搜索框就可以搜索到 jar包里的类了</p> <p>https://segmentfault.com/a/1190000016643285</p> <p>https://blog.csdn.net/qq_18581221/article/details/89320230</p></div> <footer class="page-edit" style="display:none;"><!----> <!----></footer> <!----> <!----> <!----></main> <!----></div></div></div></div><div class="global-ui"><div class="back-to-ceiling" style="right:1rem;bottom:6rem;width:2.5rem;height:2.5rem;border-radius:.25rem;line-height:2.5rem;display:none;" data-v-db14854a data-v-db14854a><svg t="1574745035067" viewBox="0 0 1024 1024" version="1.1" xmlns="http://www.w3.org/2000/svg" p-id="5404" class="icon" data-v-db14854a><path d="M526.60727968 10.90185116a27.675 27.675 0 0 0-29.21455937 0c-131.36607665 82.28402758-218.69155461 228.01873535-218.69155402 394.07834331a462.20625001 462.20625001 0 0 0 5.36959153 69.94390903c1.00431239 6.55289093-0.34802892 13.13561351-3.76865779 18.80351572-32.63518765 54.11355614-51.75690182 118.55860487-51.7569018 187.94566865a371.06718723 371.06718723 0 0 0 11.50484808 91.98906777c6.53300375 25.50556257 41.68394495 28.14064038 52.69160883 4.22606766 17.37162448-37.73630017 42.14135425-72.50938081 72.80769204-103.21549295 2.18761121 3.04276886 4.15646224 6.24463696 6.40373557 9.22774369a1871.4375 1871.4375 0 0 0 140.04691725 5.34970492 1866.36093723 1866.36093723 0 0 0 140.04691723-5.34970492c2.24727335-2.98310674 4.21612437-6.18497483 6.3937923-9.2178004 30.66633723 30.70611158 55.4360664 65.4791928 72.80769147 103.21549355 11.00766384 23.91457269 46.15860503 21.27949489 52.69160879-4.22606768a371.15156223 371.15156223 0 0 0 11.514792-91.99901164c0-69.36717486-19.13165746-133.82216804-51.75690182-187.92578088-3.42062944-5.66790279-4.76302748-12.26056868-3.76865837-18.80351632a462.20625001 462.20625001 0 0 0 5.36959269-69.943909c-0.00994388-166.08943902-87.32547796-311.81420293-218.6915546-394.09823051zM605.93803103 357.87693858a93.93749974 93.93749974 0 1 1-187.89594924 6.1e-7 93.93749974 93.93749974 0 0 1 187.89594924-6.1e-7z" p-id="5405" data-v-db14854a></path><path d="M429.50777625 765.63860547C429.50777625 803.39355007 466.44236686 1000.39046097 512.00932183 1000.39046097c45.56695499 0 82.4922232-197.00623328 82.5015456-234.7518555 0-37.75494459-36.9345906-68.35043303-82.4922232-68.34111062-45.57627738-0.00932239-82.52019037 30.59548842-82.51086798 68.34111062z" p-id="5406" data-v-db14854a></path></svg></div><!----></div></div>
    <script src="/assets/js/app.447d4224.js" defer></script><script src="/assets/js/3.9d76740c.js" defer></script><script src="/assets/js/1.c4fd7d2e.js" defer></script><script src="/assets/js/154.e090b491.js" defer></script>
  </body>
</html>
